Add queue

backend/src/ee/services/pki-acme/pki-acme-challenge-service.ts

@@ -31,7 +31,7 @@ export const pkiAcmeChallengeServiceFactory = ({
   acmeChallengeDAL
 }: TPkiAcmeChallengeServiceFactoryDep): TPkiAcmeChallengeServiceFactory => {
   const appCfg = getConfig();
-  const markChallengeAsyReady = async (challengeId: string): Promise<TPkiAcmeChallenges> => {
+  const markChallengeAsReady = async (challengeId: string): Promise<TPkiAcmeChallenges> => {
     return acmeChallengeDAL.transaction(async (tx) => {
       logger.info({ challengeId }, "Validating ACME challenge response");
       const challenge = await acmeChallengeDAL.findByIdForChallengeValidation(challengeId, tx);
@@ -65,8 +65,8 @@ export const pkiAcmeChallengeServiceFactory = ({
     });
   };
 
-  const validateChallengeResponse = async (challengeId: string): Promise<void> => {
+  const validateChallengeResponse = async (challengeId: string, retryCount: number): Promise<void> => {
-    logger.info({ challengeId }, "Validating ACME challenge response");
+    logger.info({ challengeId, retryCount }, "Validating ACME challenge response");
     const challenge = await acmeChallengeDAL.findByIdForChallengeValidation(challengeId);
     if (!challenge) {
       throw new NotFoundError({ message: "ACME challenge not found" });
@@ -93,7 +93,7 @@ export const pkiAcmeChallengeServiceFactory = ({
       //         a long running operation for long time. But assuming we are not performing a tons of
       //         challenge validation at the same time, it should be fine.
       const challengeResponse = await axios.get<string>(challengeUrl.toString(), {
-        // In case if we override the host in the development mode, still provide the original host in the header
+        // In case if we override the hos2 in the development mode, still provide the original host in the header
         // to help the upstream server to validate the request
         headers: { Host: challenge.auth.identifierValue },
         timeout: timeoutMs,
@@ -113,8 +113,10 @@ export const pkiAcmeChallengeServiceFactory = ({
       }
       await acmeChallengeDAL.markAsValidCascadeById(challengeId);
     } catch (exp) {
-      // TODO: we should retry the challenge validation a few times, but let's keep it simple for now
+      if (retryCount >= 2) {
-      await acmeChallengeDAL.markAsInvalidCascadeById(challengeId);
+        // This is the last attempt to validate the challenge response, if it fails, we mark the challenge as invalid
+        await acmeChallengeDAL.markAsInvalidCascadeById(challengeId);
+      }
       // Properly type and inspect the error
       if (axios.isAxiosError(exp)) {
         const axiosError = exp as AxiosError;
@@ -143,5 +145,5 @@ export const pkiAcmeChallengeServiceFactory = ({
     }
   };
 
-  return { markChallengeAsyReady, validateChallengeResponse };
+  return { markChallengeAsReady, validateChallengeResponse };
 };

backend/src/ee/services/pki-acme/pki-acme-queue.ts

@@ -0,0 +1,67 @@
+import { getConfig } from "@app/lib/config/env";
+import { logger } from "@app/lib/logger";
+import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+
+import { TPkiAcmeChallengeServiceFactory } from "./pki-acme-types";
+
+type TPkiAcmeQueueServiceFactoryDep = {
+  queueService: TQueueServiceFactory;
+  acmeChallengeService: TPkiAcmeChallengeServiceFactory;
+};
+
+export const pkiAcmeQueueServiceFactory = async ({
+  queueService,
+  acmeChallengeService
+}: TPkiAcmeQueueServiceFactoryDep) => {
+  const appCfg = getConfig();
+
+  // Initialize the worker to process challenge validation jobs
+  await queueService.startPg<QueueName.PkiAcmeChallengeValidation>(
+    QueueJobs.PkiAcmeChallengeValidation,
+    async ([job]) => {
+      const { challengeId } = job.data;
+      const retryCount = job.retryCount || 0;
+      try {
+        logger.info({ challengeId, retryCount }, "Processing ACME challenge validation job");
+        await acmeChallengeService.validateChallengeResponse(challengeId, retryCount);
+        logger.info({ challengeId, retryCount }, "ACME challenge validation completed successfully");
+      } catch (error) {
+        const errorMessage = error instanceof Error ? error.message : String(error);
+        logger.error(
+          error,
+          `Failed to validate ACME challenge ${challengeId} (retryCount ${retryCount}): ${errorMessage}`
+        );
+        // Re-throw to let pg-boss handle retries with exponential backoff
+        throw error;
+      }
+    },
+    {
+      batchSize: 1,
+      workerCount: 2,
+      pollingIntervalSeconds: 1
+    }
+  );
+
+  const queueChallengeValidation = async (challengeId: string): Promise<void> => {
+    if (appCfg.isSecondaryInstance) {
+      return;
+    }
+
+    logger.info({ challengeId }, "Queueing ACME challenge validation");
+    await queueService.queuePg(
+      QueueJobs.PkiAcmeChallengeValidation,
+      { challengeId },
+      {
+        retryLimit: 3,
+        retryDelay: 30 * 1000, // Base delay of 30 seconds
+        retryBackoff: true // Exponential backoff: 30s, 60s, 120s
+      }
+    );
+  };
+
+  return {
+    queueChallengeValidation
+  };
+};
+
+export type TPkiAcmeQueueServiceFactory = Awaited<ReturnType<typeof pkiAcmeQueueServiceFactory>>;

backend/src/ee/services/pki-acme/pki-acme-service.ts

@@ -62,6 +62,7 @@ import {
 import { buildUrl, extractAccountIdFromKid, validateDnsIdentifier } from "./pki-acme-fns";
 import { TPkiAcmeOrderAuthDALFactory } from "./pki-acme-order-auth-dal";
 import { TPkiAcmeOrderDALFactory } from "./pki-acme-order-dal";
+import { TPkiAcmeQueueServiceFactory } from "./pki-acme-queue";
 import {
   AcmeAuthStatus,
   AcmeChallengeStatus,
@@ -126,7 +127,8 @@ type TPkiAcmeServiceFactoryDep = {
   >;
   licenseService: Pick<TLicenseServiceFactory, "getPlan">;
   certificateV3Service: Pick<TCertificateV3ServiceFactory, "signCertificateFromProfile">;
-  acmeChallengeService: TPkiAcmeChallengeServiceFactory;
+  acmeChallengeService: Pick<TPkiAcmeChallengeServiceFactory, "markChallengeAsReady">;
+  pkiAcmeQueueService: Pick<TPkiAcmeQueueServiceFactory, "queueChallengeValidation">;
 };
 
 export const pkiAcmeServiceFactory = ({
@@ -147,7 +149,8 @@ export const pkiAcmeServiceFactory = ({
   kmsService,
   licenseService,
   certificateV3Service,
-  acmeChallengeService
+  acmeChallengeService,
+  pkiAcmeQueueService
 }: TPkiAcmeServiceFactoryDep): TPkiAcmeServiceFactory => {
   const validateAcmeProfile = async (profileId: string): Promise<TCertificateProfileWithConfigs> => {
     const profile = await certificateProfileDAL.findByIdWithConfigs(profileId);
@@ -975,7 +978,8 @@ export const pkiAcmeServiceFactory = ({
     if (!result) {
       throw new NotFoundError({ message: "ACME challenge not found" });
     }
-    await acmeChallengeService.validateChallengeResponse(challengeId);
+    await acmeChallengeService.markChallengeAsyReady(challengeId);
+    await pkiAcmeQueueService.queueChallengeValidation(challengeId);
     const challenge = (await acmeChallengeDAL.findByIdForChallengeValidation(challengeId))!;
     return {
       status: 200,

backend/src/ee/services/pki-acme/pki-acme-types.ts

@@ -178,6 +178,6 @@ export type TPkiAcmeServiceFactory = {
 };
 
 export type TPkiAcmeChallengeServiceFactory = {
-  markChallengeAsyReady: (challengeId: string) => Promise<TPkiAcmeChallenges>;
+  markChallengeAsReady: (challengeId: string) => Promise<TPkiAcmeChallenges>;
-  validateChallengeResponse: (challengeId: string) => Promise<void>;
+  validateChallengeResponse: (challengeId: string, retryCount: number) => Promise<void>;
 };

backend/src/queue/queue-service.ts

@@ -81,7 +81,8 @@ export enum QueueName {
   UserNotification = "user-notification",
   HealthAlert = "health-alert",
   CertificateV3AutoRenewal = "certificate-v3-auto-renewal",
-  PamAccountRotation = "pam-account-rotation"
+  PamAccountRotation = "pam-account-rotation",
+  PkiAcmeChallengeValidation = "pki-acme-challenge-validation"
 }
 
 export enum QueueJobs {
@@ -134,7 +135,8 @@ export enum QueueJobs {
   UserNotification = "user-notification-job",
   HealthAlert = "health-alert",
   CertificateV3DailyAutoRenewal = "certificate-v3-daily-auto-renewal",
-  PamAccountRotation = "pam-account-rotation"
+  PamAccountRotation = "pam-account-rotation",
+  PkiAcmeChallengeValidation = "pki-acme-challenge-validation"
 }
 
 export type TQueueJobTypes = {
@@ -385,6 +387,10 @@ export type TQueueJobTypes = {
     name: QueueJobs.PamAccountRotation;
     payload: undefined;
   };
+  [QueueName.PkiAcmeChallengeValidation]: {
+    name: QueueJobs.PkiAcmeChallengeValidation;
+    payload: { challengeId: string };
+  };
 };
 
 const SECRET_SCANNING_JOBS = [

backend/src/server/routes/index.ts

@@ -374,6 +374,7 @@ import { initializeOauthConfigSync } from "./v1/sso-router";
 import { registerV2Routes } from "./v2";
 import { registerV3Routes } from "./v3";
 import { registerV4Routes } from "./v4";
+import { pkiAcmeQueueServiceFactory } from "@app/ee/services/pki-acme/pki-acme-queue";
 
 const histogram = monitorEventLoopDelay({ resolution: 20 });
 histogram.enable();
@@ -2254,6 +2255,12 @@ export const registerRoutes = async (
   const acmeChallengeService = pkiAcmeChallengeServiceFactory({
     acmeChallengeDAL
   });
+
+  const pkiAcmeQueueService = pkiAcmeQueueServiceFactory({
+    queueService,
+    acmeChallengeService
+  });
+
   const pkiAcmeService = pkiAcmeServiceFactory({
     projectDAL,
     appConnectionDAL,
@@ -2272,7 +2279,8 @@ export const registerRoutes = async (
     kmsService,
     licenseService,
     certificateV3Service,
-    acmeChallengeService
+    acmeChallengeService,
+    pkiAcmeQueueService
   });
 
   const pkiSubscriberService = pkiSubscriberServiceFactory({