diff --git a/backend/src/keystore/keystore.ts b/backend/src/keystore/keystore.ts index 777e73fe2..5acfdc65e 100644 --- a/backend/src/keystore/keystore.ts +++ b/backend/src/keystore/keystore.ts @@ -38,6 +38,7 @@ export const KeyStorePrefixes = { SyncSecretIntegrationLastRunTimestamp: (projectId: string, environmentSlug: string, secretPath: string) => `sync-integration-last-run-${projectId}-${environmentSlug}-${secretPath}` as const, SecretSyncLock: (syncId: string) => `secret-sync-mutex-${syncId}` as const, + AppConnectionConcurrentJobs: (connectionId: string) => `app-connection-concurrency-${connectionId}` as const, SecretRotationLock: (rotationId: string) => `secret-rotation-v2-mutex-${rotationId}` as const, SecretScanningLock: (dataSourceId: string, resourceExternalId: string) => `secret-scanning-v2-mutex-${dataSourceId}-${resourceExternalId}` as const, diff --git a/backend/src/services/secret-sync/aws-parameter-store/aws-parameter-store-sync-fns.ts b/backend/src/services/secret-sync/aws-parameter-store/aws-parameter-store-sync-fns.ts index 680dd4256..9bcf659aa 100644 --- a/backend/src/services/secret-sync/aws-parameter-store/aws-parameter-store-sync-fns.ts +++ b/backend/src/services/secret-sync/aws-parameter-store/aws-parameter-store-sync-fns.ts @@ -12,7 +12,7 @@ type TAWSParameterStoreRecord = Record; type TAWSParameterStoreMetadataRecord = Record; type TAWSParameterStoreTagsRecord = Record>; -const MAX_RETRIES = 5; +const MAX_RETRIES = 10; const BATCH_SIZE = 10; const getSSM = async (secretSync: TAwsParameterStoreSyncWithCredentials) => { diff --git a/backend/src/services/secret-sync/aws-secrets-manager/aws-secrets-manager-sync-fns.ts b/backend/src/services/secret-sync/aws-secrets-manager/aws-secrets-manager-sync-fns.ts index 8e37cf277..8573d6ffc 100644 --- a/backend/src/services/secret-sync/aws-secrets-manager/aws-secrets-manager-sync-fns.ts +++ b/backend/src/services/secret-sync/aws-secrets-manager/aws-secrets-manager-sync-fns.ts @@ -38,7 +38,7 @@ type TAwsSecretsRecord = Record; type TAwsSecretValuesRecord = Record; type TAwsSecretDescriptionsRecord = Record; -const MAX_RETRIES = 5; +const MAX_RETRIES = 10; const BATCH_SIZE = 20; const getSecretsManagerClient = async (secretSync: TAwsSecretsManagerSyncWithCredentials) => { diff --git a/backend/src/services/secret-sync/secret-sync-queue.ts b/backend/src/services/secret-sync/secret-sync-queue.ts index 7bef7d8c7..797e1dc1f 100644 --- a/backend/src/services/secret-sync/secret-sync-queue.ts +++ b/backend/src/services/secret-sync/secret-sync-queue.ts @@ -81,7 +81,7 @@ type TSecretSyncQueueFactoryDep = { | "invalidateSecretCacheByProjectId" >; secretImportDAL: Pick; - secretSyncDAL: Pick; + secretSyncDAL: Pick; auditLogService: Pick; projectMembershipDAL: Pick; projectDAL: TProjectDALFactory; @@ -104,17 +104,15 @@ type SecretSyncActionJob = Job< TQueueSecretSyncSyncSecretsByIdDTO | TQueueSecretSyncImportSecretsByIdDTO | TQueueSecretSyncRemoveSecretsByIdDTO >; +const JITTER_MS = 10 * 1000; +const REQUEUE_MS = 30 * 1000; +const REQUEUE_LIMIT = 30; +const CONNECTION_CONCURRENCY_LIMIT = 3; + const getRequeueDelay = (failureCount?: number) => { - if (!failureCount) return 0; - - const baseDelay = 1000; - const maxDelay = 30000; - - const delay = Math.min(baseDelay * 2 ** failureCount, maxDelay); - - const jitter = delay * (0.5 + Math.random() * 0.5); - - return jitter; + const jitter = Math.random() * JITTER_MS; + if (!failureCount) return jitter; + return REQUEUE_MS + jitter; }; export const secretSyncQueueFactory = ({ @@ -193,6 +191,46 @@ export const secretSyncQueueFactory = ({ folderCommitService }); + const $isConnectionConcurrencyLimitReached = async (connectionId: string) => { + const concurrencyCount = await keyStore.getItem(KeyStorePrefixes.AppConnectionConcurrentJobs(connectionId)); + + if (!concurrencyCount) return false; + + const count = Number.parseInt(concurrencyCount, 10); + + if (Number.isNaN(count)) return false; + + return count >= CONNECTION_CONCURRENCY_LIMIT; + }; + + const $incrementConnectionConcurrencyCount = async (connectionId: string) => { + const concurrencyCount = await keyStore.getItem(KeyStorePrefixes.AppConnectionConcurrentJobs(connectionId)); + + const currentCount = Number.parseInt(concurrencyCount || "0", 10); + + const incrementedCount = Number.isNaN(currentCount) ? 1 : currentCount + 1; + + await keyStore.setItemWithExpiry( + KeyStorePrefixes.AppConnectionConcurrentJobs(connectionId), + (REQUEUE_MS * REQUEUE_LIMIT) / 1000, // in seconds + incrementedCount + ); + }; + + const $decrementConnectionConcurrencyCount = async (connectionId: string) => { + const concurrencyCount = await keyStore.getItem(KeyStorePrefixes.AppConnectionConcurrentJobs(connectionId)); + + const currentCount = Number.parseInt(concurrencyCount || "0", 10); + + const decrementedCount = Math.max(0, Number.isNaN(currentCount) ? 0 : currentCount - 1); + + await keyStore.setItemWithExpiry( + KeyStorePrefixes.AppConnectionConcurrentJobs(connectionId), + (REQUEUE_MS * REQUEUE_LIMIT) / 1000, // in seconds + decrementedCount + ); + }; + const $getInfisicalSecrets = async ( secretSync: TSecretSyncRaw | TSecretSyncWithCredentials, includeImports = true @@ -416,15 +454,11 @@ export const secretSyncQueueFactory = ({ return importedSecretMap; }; - const $handleSyncSecretsJob = async (job: TSecretSyncSyncSecretsDTO) => { + const $handleSyncSecretsJob = async (job: TSecretSyncSyncSecretsDTO, secretSync: TSecretSyncRaw) => { const { data: { syncId, auditLogInfo } } = job; - const secretSync = await secretSyncDAL.findById(syncId); - - if (!secretSync) throw new Error(`Cannot find secret sync with ID ${syncId}`); - await enterpriseSyncCheck( licenseService, secretSync.destination as SecretSync, @@ -566,15 +600,11 @@ export const secretSyncQueueFactory = ({ logger.info("SecretSync Sync Job with ID %s Completed", job.id); }; - const $handleImportSecretsJob = async (job: TSecretSyncImportSecretsDTO) => { + const $handleImportSecretsJob = async (job: TSecretSyncImportSecretsDTO, secretSync: TSecretSyncRaw) => { const { data: { syncId, auditLogInfo, importBehavior } } = job; - const secretSync = await secretSyncDAL.findById(syncId); - - if (!secretSync) throw new Error(`Cannot find secret sync with ID ${syncId}`); - await secretSyncDAL.updateById(syncId, { importStatus: SecretSyncStatus.Running }); @@ -683,15 +713,11 @@ export const secretSyncQueueFactory = ({ logger.info("SecretSync Import Job with ID %s Completed", job.id); }; - const $handleRemoveSecretsJob = async (job: TSecretSyncRemoveSecretsDTO) => { + const $handleRemoveSecretsJob = async (job: TSecretSyncRemoveSecretsDTO, secretSync: TSecretSyncRaw) => { const { data: { syncId, auditLogInfo, deleteSyncOnComplete } } = job; - const secretSync = await secretSyncDAL.findById(syncId); - - if (!secretSync) throw new Error(`Cannot find secret sync with ID ${syncId}`); - await enterpriseSyncCheck( licenseService, secretSync.destination as SecretSync, @@ -894,6 +920,17 @@ export const secretSyncQueueFactory = ({ const secretSyncs = await secretSyncDAL.find({ folderId: folder.id, isAutoSyncEnabled: true }); + await secretSyncDAL.update( + { + $in: { + id: secretSyncs.map((sync) => sync.id) + } + }, + { + syncStatus: SecretSyncStatus.Pending + } + ); + await Promise.all(secretSyncs.map((secretSync) => queueSecretSyncSyncSecretsById({ syncId: secretSync.id }))); }; @@ -904,7 +941,7 @@ export const secretSyncQueueFactory = ({ case QueueJobs.SecretSyncSyncSecrets: { const { failedToAcquireLockCount = 0, ...rest } = job.data as TQueueSecretSyncSyncSecretsByIdDTO; - if (failedToAcquireLockCount < 10) { + if (failedToAcquireLockCount < REQUEUE_LIMIT) { await queueSecretSyncSyncSecretsById({ ...rest, failedToAcquireLockCount: failedToAcquireLockCount + 1 }); return; } @@ -974,6 +1011,26 @@ export const secretSyncQueueFactory = ({ | TQueueSecretSyncImportSecretsByIdDTO | TQueueSecretSyncRemoveSecretsByIdDTO; + const secretSync = await secretSyncDAL.findById(syncId); + + if (!secretSync) throw new Error(`Cannot find secret sync with ID ${syncId}`); + + const { connectionId } = secretSync; + + if (job.name === QueueJobs.SecretSyncSyncSecrets) { + const isConcurrentLimitReached = await $isConnectionConcurrencyLimitReached(connectionId); + + if (isConcurrentLimitReached) { + logger.info( + `SecretSync Concurrency limit reached [syncId=${syncId}] [job=${job.name}] [connectionId=${connectionId}]` + ); + + await $handleAcquireLockFailure(job as SecretSyncActionJob); + + return; + } + } + let lock: Awaited>; try { @@ -993,20 +1050,26 @@ export const secretSyncQueueFactory = ({ try { switch (job.name) { - case QueueJobs.SecretSyncSyncSecrets: - await $handleSyncSecretsJob(job as TSecretSyncSyncSecretsDTO); + case QueueJobs.SecretSyncSyncSecrets: { + await $incrementConnectionConcurrencyCount(connectionId); + await $handleSyncSecretsJob(job as TSecretSyncSyncSecretsDTO, secretSync); break; + } case QueueJobs.SecretSyncImportSecrets: - await $handleImportSecretsJob(job as TSecretSyncImportSecretsDTO); + await $handleImportSecretsJob(job as TSecretSyncImportSecretsDTO, secretSync); break; case QueueJobs.SecretSyncRemoveSecrets: - await $handleRemoveSecretsJob(job as TSecretSyncRemoveSecretsDTO); + await $handleRemoveSecretsJob(job as TSecretSyncRemoveSecretsDTO, secretSync); break; default: // eslint-disable-next-line @typescript-eslint/restrict-template-expressions throw new Error(`Unhandled Secret Sync Job ${job.name}`); } } finally { + if (job.name === QueueJobs.SecretSyncSyncSecrets) { + await $decrementConnectionConcurrencyCount(connectionId); + } + await lock.release(); } }); diff --git a/frontend/src/components/secret-syncs/SecretSyncStatusBadge.tsx b/frontend/src/components/secret-syncs/SecretSyncStatusBadge.tsx index 53b53d5c0..e0d541138 100644 --- a/frontend/src/components/secret-syncs/SecretSyncStatusBadge.tsx +++ b/frontend/src/components/secret-syncs/SecretSyncStatusBadge.tsx @@ -1,6 +1,7 @@ import { faCheck, faExclamationTriangle, + faHourglass, faRotate, IconDefinition } from "@fortawesome/free-solid-svg-icons"; @@ -29,7 +30,11 @@ export const SecretSyncStatusBadge = ({ status }: Props) => { text = "Synced"; icon = faCheck; break; - case SecretSyncStatus.Pending: // no need to differentiate from user perspective + case SecretSyncStatus.Pending: + variant = "primary"; + text = "Queued"; + icon = faHourglass; + break; case SecretSyncStatus.Running: default: variant = "primary"; @@ -42,11 +47,7 @@ export const SecretSyncStatusBadge = ({ status }: Props) => { {text}