🐞 Fix startup freeze in EmailQueueService during Redis connection

apps/web/src/instrumentation.ts

@@ -13,18 +13,29 @@ export async function register() {
   if (process.env.NEXT_RUNTIME === "nodejs" && !initialized) {
     console.log("Registering instrumentation");
 
-    const { EmailQueueService } = await import(
-      "~/server/service/email-queue-service"
-    );
-    await EmailQueueService.init();
+    try {
+      const { EmailQueueService } = await import(
+        "~/server/service/email-queue-service"
+      );
+
+      console.log("Initializing EmailQueueService...");
+      await EmailQueueService.init();
+      console.log("EmailQueueService initialized successfully");
 
-    /**
-     * Send usage data to Stripe
-     */
-    if (isCloud()) {
-      await import("~/server/jobs/usage-job");
-    }
+      /**
+       * Send usage data to Stripe
+       */
+      if (isCloud()) {
+        await import("~/server/jobs/usage-job");
+      }
 
-    initialized = true;
+      initialized = true;
+      console.log("Instrumentation registration completed");
+    } catch (error) {
+      console.error("Failed to initialize services during instrumentation:", error);
+      // Don't throw here to prevent application from completely failing to start
+      // The error will be logged and the application can still handle requests
+      // Individual services will handle their own initialization as needed
+    }
   }
 }

apps/web/src/server/redis.ts

@@ -7,6 +7,23 @@ export const getRedis = () => {
   if (!connection) {
     connection = new IORedis(`${env.REDIS_URL}?family=0`, {
       maxRetriesPerRequest: null,
+      connectTimeout: 10000, // 10 seconds
+      lazyConnect: true,
+      retryDelayOnClusterFailover: 100,
+      enableReadyCheck: false,
+    });
+
+    // Add error handler to prevent unhandled connection errors
+    connection.on('error', (error) => {
+      console.error('[Redis]: Connection error:', error);
+    });
+
+    connection.on('connect', () => {
+      console.log('[Redis]: Connected successfully');
+    });
+
+    connection.on('ready', () => {
+      console.log('[Redis]: Ready to accept commands');
     });
   }
   return connection;

apps/web/src/server/service/email-queue-service.ts

@@ -25,14 +25,41 @@ function createQueueAndWorker(region: string, quota: number, suffix: string) {
 
   const queueName = `${region}-${suffix}`;
 
-  const queue = new Queue(queueName, { connection });
+  const queue = new Queue(queueName, { 
+    connection,
+    defaultJobOptions: {
+      removeOnComplete: 100,
+      removeOnFail: 50,
+    },
+  });
 
   // TODO: Add team context to job data when queueing
   const worker = new Worker(queueName, createWorkerHandler(executeEmail), {
     concurrency: quota,
     connection,
   });
 
+  // Add error handlers to prevent unhandled errors from causing hangs
+  worker.on('error', (error) => {
+    logger.error(
+      { error, queueName, region },
+      `[EmailQueueService]: Worker error for ${queueName}`
+    );
+  });
+
+  queue.on('error', (error) => {
+    logger.error(
+      { error, queueName, region },
+      `[EmailQueueService]: Queue error for ${queueName}`
+    );
+  });
+
+  // Log successful creation
+  logger.info(
+    { queueName, region, quota },
+    `[EmailQueueService]: Successfully created queue and worker for ${queueName}`
+  );
+
   return { queue, worker };
 }
 
@@ -43,7 +70,7 @@ export class EmailQueueService {
   public static marketingQueue = new Map<string, Queue<QueueEmailJob>>();
   private static marketingWorker = new Map<string, Worker>();
 
-  public static initializeQueue(
+  public static async initializeQueue(
     region: string,
     quota: number,
     transactionalQuotaPercentage: number
@@ -58,56 +85,100 @@ export class EmailQueueService {
     );
     const marketingQuota = quota - transactionalQuota;
 
-    if (this.transactionalQueue.has(region)) {
-      logger.info(
-        { region, transactionalQuota },
-        `[EmailQueueService]: Updating transactional quota for region`
-      );
-      const transactionalWorker = this.transactionalWorker.get(region);
-      if (transactionalWorker) {
-        transactionalWorker.concurrency =
-          transactionalQuota !== 0 ? transactionalQuota : 1;
+    try {
+      if (this.transactionalQueue.has(region)) {
+        logger.info(
+          { region, transactionalQuota },
+          `[EmailQueueService]: Updating transactional quota for region`
+        );
+        const transactionalWorker = this.transactionalWorker.get(region);
+        if (transactionalWorker) {
+          transactionalWorker.concurrency =
+            transactionalQuota !== 0 ? transactionalQuota : 1;
+        }
+      } else {
+        logger.info(
+          { region, transactionalQuota },
+          `[EmailQueueService]: Creating transactional queue for region`
+        );
+
+        const { queue: transactionalQueue, worker: transactionalWorker } =
+          await this.createQueueAndWorkerWithTimeout(
+            region,
+            transactionalQuota !== 0 ? transactionalQuota : 1,
+            "transaction"
+          );
+        this.transactionalQueue.set(region, transactionalQueue);
+        this.transactionalWorker.set(region, transactionalWorker);
       }
-    } else {
-      logger.info(
-        { region, transactionalQuota },
-        `[EmailQueueService]: Creating transactional queue for region`
-      );
-      const { queue: transactionalQueue, worker: transactionalWorker } =
-        createQueueAndWorker(
-          region,
-          transactionalQuota !== 0 ? transactionalQuota : 1,
-          "transaction"
+
+      if (this.marketingQueue.has(region)) {
+        logger.info(
+          { region, marketingQuota },
+          `[EmailQueueService]: Updating marketing quota for region`
         );
-      this.transactionalQueue.set(region, transactionalQueue);
-      this.transactionalWorker.set(region, transactionalWorker);
-    }
+        const marketingWorker = this.marketingWorker.get(region);
+        if (marketingWorker) {
+          marketingWorker.concurrency = marketingQuota !== 0 ? marketingQuota : 1;
+        }
+      } else {
+        logger.info(
+          { region, marketingQuota },
+          `[EmailQueueService]: Creating marketing queue for region`
+        );
+
+        const { queue: marketingQueue, worker: marketingWorker } =
+          await this.createQueueAndWorkerWithTimeout(
+            region,
+            marketingQuota !== 0 ? marketingQuota : 1,
+            "marketing"
+          );
+        this.marketingQueue.set(region, marketingQueue);
+        this.marketingWorker.set(region, marketingWorker);
+      }
 
-    if (this.marketingQueue.has(region)) {
       logger.info(
-        { region, marketingQuota },
-        `[EmailQueueService]: Updating marketing quota for region`
+        { region },
+        `[EmailQueueService]: Successfully initialized queues for region`
       );
-      const marketingWorker = this.marketingWorker.get(region);
-      if (marketingWorker) {
-        marketingWorker.concurrency = marketingQuota !== 0 ? marketingQuota : 1;
-      }
-    } else {
-      logger.info(
-        { region, marketingQuota },
-        `[EmailQueueService]: Creating marketing queue for region`
+    } catch (error) {
+      logger.error(
+        { error, region },
+        `[EmailQueueService]: Failed to initialize queues for region`
       );
-      const { queue: marketingQueue, worker: marketingWorker } =
-        createQueueAndWorker(
-          region,
-          marketingQuota !== 0 ? marketingQuota : 1,
-          "marketing"
-        );
-      this.marketingQueue.set(region, marketingQueue);
-      this.marketingWorker.set(region, marketingWorker);
+      throw error;
     }
   }
 
+  private static async createQueueAndWorkerWithTimeout(
+    region: string,
+    quota: number,
+    suffix: string
+  ): Promise<{ queue: Queue<QueueEmailJob>; worker: Worker }> {
+    return new Promise(async (resolve, reject) => {
+      const timeout = setTimeout(() => {
+        reject(new Error(`Timeout creating queue and worker for ${region}-${suffix}`));
+      }, 15000); // 15 second timeout
+
+      try {
+        // Ensure Redis connection is ready before creating queue/worker
+        const connection = getRedis();
+
+        // Force connection if not already connected
+        if (connection.status !== 'ready' && connection.status !== 'connected') {
+          await connection.connect();
+        }
+
+        const result = createQueueAndWorker(region, quota, suffix);
+        clearTimeout(timeout);
+        resolve(result);
+      } catch (error) {
+        clearTimeout(timeout);
+        reject(error);
+      }
+    });
+  }
+
   public static async queueEmail(
     emailId: string,
     teamId: number,
@@ -293,7 +364,7 @@ export class EmailQueueService {
   public static async init() {
     const sesSettings = await db.sesSetting.findMany();
     for (const sesSetting of sesSettings) {
-      this.initializeQueue(
+      await this.initializeQueue(
         sesSetting.region,
         sesSetting.sesEmailRateLimit,
         sesSetting.transactionalQuota

apps/web/src/server/service/ses-settings-service.ts

@@ -117,7 +117,7 @@ export class SesSettingsService {
 
       await registerConfigurationSet(setting);
 
-      EmailQueueService.initializeQueue(
+      await EmailQueueService.initializeQueue(
         region,
         setting.sesEmailRateLimit,
         setting.transactionalQuota,
@@ -175,7 +175,7 @@ export class SesSettingsService {
       "Email queues before update",
     );
 
-    EmailQueueService.initializeQueue(
+    await EmailQueueService.initializeQueue(
       setting.region,
       setting.sesEmailRateLimit,
       setting.transactionalQuota,