diff --git a/backend/e2e-test/routes/v1/secret-replication.spec.ts b/backend/e2e-test/routes/v1/secret-replication.spec.ts index f762165a2..b56a8f9a0 100644 --- a/backend/e2e-test/routes/v1/secret-replication.spec.ts +++ b/backend/e2e-test/routes/v1/secret-replication.spec.ts @@ -118,9 +118,9 @@ describe.each([{ secretPath: "/" }, { secretPath: "/deep" }])( value: "stage-value" }); - // wait for 5 second for replication to finish + // wait for 10 second for replication to finish await new Promise((resolve) => { - setTimeout(resolve, 5000); // time to breathe for db + setTimeout(resolve, 10000); // time to breathe for db }); const secret = await getSecretByNameV2({ @@ -173,9 +173,9 @@ describe.each([{ secretPath: "/" }, { secretPath: "/deep" }])( value: "prod-value" }); - // wait for 5 second for replication to finish + // wait for 10 second for replication to finish await new Promise((resolve) => { - setTimeout(resolve, 5000); // time to breathe for db + setTimeout(resolve, 10000); // time to breathe for db }); const secret = await getSecretByNameV2({ @@ -343,9 +343,9 @@ describe.each([{ path: "/" }, { path: "/deep" }])( value: "prod-value" }); - // wait for 5 second for replication to finish + // wait for 10 second for replication to finish await new Promise((resolve) => { - setTimeout(resolve, 5000); // time to breathe for db + setTimeout(resolve, 10000); // time to breathe for db }); const secret = await getSecretByNameV2({ diff --git a/backend/src/keystore/keystore.ts b/backend/src/keystore/keystore.ts index 5b2c7fba9..723a22817 100644 --- a/backend/src/keystore/keystore.ts +++ b/backend/src/keystore/keystore.ts @@ -29,7 +29,7 @@ export const KeyStorePrefixes = { }; export const KeyStoreTtls = { - SetSyncSecretIntegrationLastRunTimestampInSeconds: 10, + SetSyncSecretIntegrationLastRunTimestampInSeconds: 60, AccessTokenStatusUpdateInSeconds: 120 }; diff --git a/backend/src/services/secret/secret-queue.ts b/backend/src/services/secret/secret-queue.ts index 4b41144b4..3d75fa4f8 100644 --- a/backend/src/services/secret/secret-queue.ts +++ b/backend/src/services/secret/secret-queue.ts @@ -112,6 +112,8 @@ export type TGetSecrets = { }; const MAX_SYNC_SECRET_DEPTH = 5; +const SYNC_SECRET_DEBOUNCE_INTERVAL_MS = 3000; + export const uniqueSecretQueueKey = (environment: string, secretPath: string) => `secret-queue-dedupe-${environment}-${secretPath}`; @@ -169,6 +171,39 @@ export const secretQueueFactory = ({ ); }; + const $generateActor = async (actorId?: string, isManual?: boolean): Promise => { + if (isManual && actorId) { + const user = await userDAL.findById(actorId); + + if (!user) { + throw new Error("User not found"); + } + + return { + type: ActorType.USER, + metadata: { + email: user.email, + username: user.username, + userId: user.id + } + }; + } + + return { + type: ActorType.PLATFORM, + metadata: {} + }; + }; + + const $getJobKey = (projectId: string, environmentSlug: string, secretPath: string) => { + // the idea here is a timestamp based id which will be constant in a 3s interval + const timestampId = Math.floor(Date.now() / SYNC_SECRET_DEBOUNCE_INTERVAL_MS); + + return `secret-queue-sync_${projectId}_${environmentSlug}_${secretPath}_${timestampId}` + .replace("/", "-") + .replace(":", "-"); + }; + const addSecretReminder = async ({ oldSecret, newSecret, projectId }: TCreateSecretReminderDTO) => { try { const appCfg = getConfig(); @@ -466,7 +501,7 @@ export const secretQueueFactory = ({ dto: TGetSecrets & { isManual?: boolean; actorId?: string; deDupeQueue?: Record } ) => { await queueService.queue(QueueName.IntegrationSync, QueueJobs.IntegrationSync, dto, { - attempts: 3, + attempts: 5, delay: 1000, backoff: { type: "exponential", @@ -479,10 +514,10 @@ export const secretQueueFactory = ({ const replicateSecrets = async (dto: Omit) => { await queueService.queue(QueueName.SecretReplication, QueueJobs.SecretReplication, dto, { - attempts: 3, + attempts: 5, backoff: { type: "exponential", - delay: 2000 + delay: 3000 }, removeOnComplete: true, removeOnFail: true @@ -499,6 +534,7 @@ export const secretQueueFactory = ({ logger.info( `syncSecrets: syncing project secrets where [projectId=${dto.projectId}] [environment=${dto.environmentSlug}] [path=${dto.secretPath}]` ); + const deDuplicationKey = uniqueSecretQueueKey(dto.environmentSlug, dto.secretPath); if ( !dto.excludeReplication @@ -523,7 +559,8 @@ export const secretQueueFactory = ({ { removeOnFail: true, removeOnComplete: true, - delay: 1000, + jobId: $getJobKey(dto.projectId, dto.environmentSlug, dto.secretPath), + delay: SYNC_SECRET_DEBOUNCE_INTERVAL_MS, attempts: 5, backoff: { type: "exponential", @@ -532,7 +569,6 @@ export const secretQueueFactory = ({ } ); }; - const sendFailedIntegrationSyncEmails = async (payload: TFailedIntegrationSyncEmailsPayload) => { const appCfg = getConfig(); if (!appCfg.isSmtpConfigured) return; @@ -540,7 +576,6 @@ export const secretQueueFactory = ({ await queueService.queue(QueueName.IntegrationSync, QueueJobs.SendFailedIntegrationSyncEmails, payload, { jobId: `send-failed-integration-sync-emails-${payload.projectId}-${payload.secretPath}-${payload.environmentSlug}`, delay: 1_000 * 60, // 1 minute - removeOnFail: true, removeOnComplete: true }); @@ -733,80 +768,51 @@ export const secretQueueFactory = ({ ); } - const integrations = await integrationDAL.findByProjectIdV2(projectId, environment); // note: returns array of integrations + integration auths in this environment - const toBeSyncedIntegrations = integrations.filter( - // note: sync only the integrations sourced from secretPath - ({ secretPath: integrationSecPath, isActive }) => isActive && isSamePath(secretPath, integrationSecPath) + const lock = await keyStore.acquireLock( + [KeyStorePrefixes.SyncSecretIntegrationLock(projectId, environment, secretPath)], + 60000, + { + retryCount: 10, + retryDelay: 3000, + retryJitter: 500 + } ); const integrationsFailedToSync: { integrationId: string; syncMessage?: string }[] = []; - - if (!integrations.length) return; - logger.info( - `getIntegrationSecrets: secret integration sync started [jobId=${job.id}] [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${environment}] [secretPath=${job.data.secretPath}] [depth=${depth}]` - ); - - const lock = await keyStore.acquireLock( - [KeyStorePrefixes.SyncSecretIntegrationLock(projectId, environment, secretPath)], - 10000, - { - retryCount: 3, - retryDelay: 2000 - } - ); const lockAcquiredTime = new Date(); - const lastRunSyncIntegrationTimestamp = await keyStore.getItem( - KeyStorePrefixes.SyncSecretIntegrationLastRunTimestamp(projectId, environment, secretPath) - ); - - // check whether the integration should wait or not - if (lastRunSyncIntegrationTimestamp) { - const INTEGRATION_INTERVAL = 2000; - const isStaleSyncIntegration = new Date(job.timestamp) < new Date(lastRunSyncIntegrationTimestamp); - if (isStaleSyncIntegration) { - logger.info( - `getIntegrationSecrets: secret integration sync stale [jobId=${job.id}] [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${environment}] [secretPath=${job.data.secretPath}] [depth=${depth}]` - ); - return; - } - - const timeDifferenceWithLastIntegration = getTimeDifferenceInSeconds( - lockAcquiredTime.toISOString(), - lastRunSyncIntegrationTimestamp - ); - if (timeDifferenceWithLastIntegration < INTEGRATION_INTERVAL && timeDifferenceWithLastIntegration > 0) - await new Promise((resolve) => { - setTimeout(resolve, 2000 - timeDifferenceWithLastIntegration * 1000); - }); - } - - const generateActor = async (): Promise => { - if (isManual && actorId) { - const user = await userDAL.findById(actorId); - - if (!user) { - throw new Error("User not found"); - } - - return { - type: ActorType.USER, - metadata: { - email: user.email, - username: user.username, - userId: user.id - } - }; - } - - return { - type: ActorType.PLATFORM, - metadata: {} - }; - }; - // akhilmhdh: this try catch is for lock release try { + const lastRunSyncIntegrationTimestamp = await keyStore.getItem( + KeyStorePrefixes.SyncSecretIntegrationLastRunTimestamp(projectId, environment, secretPath) + ); + + // check whether the integration should wait or not + if (lastRunSyncIntegrationTimestamp) { + const INTEGRATION_INTERVAL = 2000; + + const timeDifferenceWithLastIntegration = getTimeDifferenceInSeconds( + lockAcquiredTime.toISOString(), + lastRunSyncIntegrationTimestamp + ); + + // give some time for integration to breath + if (timeDifferenceWithLastIntegration < INTEGRATION_INTERVAL) + await new Promise((resolve) => { + setTimeout(resolve, INTEGRATION_INTERVAL); + }); + } + + const integrations = await integrationDAL.findByProjectIdV2(projectId, environment); // note: returns array of integrations + integration auths in this environment + const toBeSyncedIntegrations = integrations.filter( + // note: sync only the integrations sourced from secretPath + ({ secretPath: integrationSecPath, isActive }) => isActive && isSamePath(secretPath, integrationSecPath) + ); + + if (!integrations.length) return; + logger.info( + `getIntegrationSecrets: secret integration sync started [jobId=${job.id}] [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${environment}] [secretPath=${job.data.secretPath}] [depth=${depth}]` + ); const secrets = shouldUseSecretV2Bridge ? await getIntegrationSecretsV2({ environment, @@ -892,7 +898,7 @@ export const secretQueueFactory = ({ await auditLogService.createAuditLog({ projectId, - actor: await generateActor(), + actor: await $generateActor(actorId, isManual), event: { type: EventType.INTEGRATION_SYNCED, metadata: { @@ -931,7 +937,7 @@ export const secretQueueFactory = ({ await auditLogService.createAuditLog({ projectId, - actor: await generateActor(), + actor: await $generateActor(actorId, isManual), event: { type: EventType.INTEGRATION_SYNCED, metadata: {