From da7cc1e406e5b66ab6cfc548275b0c4090bebfcc Mon Sep 17 00:00:00 2001 From: Fang-Pen Lin Date: Mon, 1 Dec 2025 15:09:48 -0800 Subject: [PATCH] Add queue --- .../pki-acme/pki-acme-challenge-service.ts | 16 +++-- .../ee/services/pki-acme/pki-acme-queue.ts | 67 +++++++++++++++++++ .../ee/services/pki-acme/pki-acme-service.ts | 10 ++- .../ee/services/pki-acme/pki-acme-types.ts | 4 +- backend/src/queue/queue-service.ts | 10 ++- backend/src/server/routes/index.ts | 10 ++- 6 files changed, 102 insertions(+), 15 deletions(-) create mode 100644 backend/src/ee/services/pki-acme/pki-acme-queue.ts diff --git a/backend/src/ee/services/pki-acme/pki-acme-challenge-service.ts b/backend/src/ee/services/pki-acme/pki-acme-challenge-service.ts index e0d58144f..572d52bf5 100644 --- a/backend/src/ee/services/pki-acme/pki-acme-challenge-service.ts +++ b/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 => { + const markChallengeAsReady = async (challengeId: string): Promise => { 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 => { - logger.info({ challengeId }, "Validating ACME challenge response"); + const validateChallengeResponse = async (challengeId: string, retryCount: number): Promise => { + 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(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 - await acmeChallengeDAL.markAsInvalidCascadeById(challengeId); + if (retryCount >= 2) { + // 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 }; }; diff --git a/backend/src/ee/services/pki-acme/pki-acme-queue.ts b/backend/src/ee/services/pki-acme/pki-acme-queue.ts new file mode 100644 index 000000000..2adef485e --- /dev/null +++ b/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( + 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 => { + 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>; diff --git a/backend/src/ee/services/pki-acme/pki-acme-service.ts b/backend/src/ee/services/pki-acme/pki-acme-service.ts index 89ba17dd5..167161475 100644 --- a/backend/src/ee/services/pki-acme/pki-acme-service.ts +++ b/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; certificateV3Service: Pick; - acmeChallengeService: TPkiAcmeChallengeServiceFactory; + acmeChallengeService: Pick; + pkiAcmeQueueService: Pick; }; export const pkiAcmeServiceFactory = ({ @@ -147,7 +149,8 @@ export const pkiAcmeServiceFactory = ({ kmsService, licenseService, certificateV3Service, - acmeChallengeService + acmeChallengeService, + pkiAcmeQueueService }: TPkiAcmeServiceFactoryDep): TPkiAcmeServiceFactory => { const validateAcmeProfile = async (profileId: string): Promise => { 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, diff --git a/backend/src/ee/services/pki-acme/pki-acme-types.ts b/backend/src/ee/services/pki-acme/pki-acme-types.ts index fb76aec79..6607ce711 100644 --- a/backend/src/ee/services/pki-acme/pki-acme-types.ts +++ b/backend/src/ee/services/pki-acme/pki-acme-types.ts @@ -178,6 +178,6 @@ export type TPkiAcmeServiceFactory = { }; export type TPkiAcmeChallengeServiceFactory = { - markChallengeAsyReady: (challengeId: string) => Promise; - validateChallengeResponse: (challengeId: string) => Promise; + markChallengeAsReady: (challengeId: string) => Promise; + validateChallengeResponse: (challengeId: string, retryCount: number) => Promise; }; diff --git a/backend/src/queue/queue-service.ts b/backend/src/queue/queue-service.ts index 57409d173..5ea6dff8e 100644 --- a/backend/src/queue/queue-service.ts +++ b/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 = [ diff --git a/backend/src/server/routes/index.ts b/backend/src/server/routes/index.ts index ec9e56880..9aea9d5c1 100644 --- a/backend/src/server/routes/index.ts +++ b/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({