From 96190bda225219a1ac277db036d503ff117e8a0f Mon Sep 17 00:00:00 2001 From: Surya Prashanth Date: Thu, 19 Jun 2025 19:08:10 +0530 Subject: [PATCH] [SILO-324] fix: handle duplicate page runs in clickup (#3461) * [SILO-324] fix: handle duplicate page runs in clickup * increase mq heartbeat to 5 minutes --- apiserver/plane/bgtasks/data_import_task.py | 3 +- .../etl/src/clickup/services/api.service.ts | 5 +++- .../migrator/clickup-additional.migrator.ts | 28 +++++++++++++++++++ .../migrator/clickup.migrator.ts | 10 ++++++- .../apps/engine/controllers/job.controller.ts | 5 +++- silo/src/worker/base/mq.ts | 2 +- 6 files changed, 48 insertions(+), 5 deletions(-) diff --git a/apiserver/plane/bgtasks/data_import_task.py b/apiserver/plane/bgtasks/data_import_task.py index ee76d10193..0d4c798c5e 100644 --- a/apiserver/plane/bgtasks/data_import_task.py +++ b/apiserver/plane/bgtasks/data_import_task.py @@ -79,7 +79,8 @@ def update_job_batch_completion( + imported_issues, errored_issue_count=Coalesce(F("errored_issue_count"), 0) + (total_issues - imported_issues), - completed_batch_count=Coalesce(F("completed_batch_count"), 0) + 1, + completed_batch_count=Coalesce(F("completed_batch_count"), 0) + + (1 if imported_batch_count > 0 else 0), updated_at=timezone.now(), ) diff --git a/packages/etl/src/clickup/services/api.service.ts b/packages/etl/src/clickup/services/api.service.ts index 552c8949ce..24f255ef01 100644 --- a/packages/etl/src/clickup/services/api.service.ts +++ b/packages/etl/src/clickup/services/api.service.ts @@ -45,8 +45,11 @@ export class ClickupAPIService { if (error.response?.status === 429) { const headers = error.response.headers as RawAxiosResponseHeaders; if (headers && headers["x-ratelimit-reset"]) { - const waitTime = getWaitTimeInMs(headers["x-ratelimit-reset"] as string); + let waitTime = getWaitTimeInMs(headers["x-ratelimit-reset"] as string); console.log("Rate limit exceeded ======> waiting for", waitTime, "ms"); + // add a random jitter of 20% to the wait time + const jitter = Math.random() * 0.2; + waitTime = waitTime * (1 + jitter); await new Promise((resolve) => setTimeout(resolve, waitTime)); return this.client.request(error.config); } diff --git a/silo/src/apps/clickup-importer/migrator/clickup-additional.migrator.ts b/silo/src/apps/clickup-importer/migrator/clickup-additional.migrator.ts index 111f4fbec0..2fbafa11c5 100644 --- a/silo/src/apps/clickup-importer/migrator/clickup-additional.migrator.ts +++ b/silo/src/apps/clickup-importer/migrator/clickup-additional.migrator.ts @@ -26,6 +26,7 @@ import { logger } from "@/logger"; import { getAPIClient } from "@/services/client"; import { TaskHandler, TaskHeaders } from "@/types"; import { MQ, Store } from "@/worker/base"; +import { Lock } from "@/worker/base/lock"; import { TBatch } from "@/worker/types"; import { getClickUpClient } from "../helpers/clickup-client"; import { ClickUpBulkTransformer } from "./transformers/etl"; @@ -67,6 +68,15 @@ export class ClickUpAdditionalDataMigrator extends TaskHandler { return false; } + // If the job has been cancelled return + if (job.cancelled_at) { + logger.info(`[${headers.route.toUpperCase()}] Job cancelled, skipping`, { + jobId: headers.jobId, + type: headers.type, + }); + return false; + } + let lastResult: any; switch (headers.type) { @@ -111,6 +121,24 @@ export class ClickUpAdditionalDataMigrator extends TaskHandler { const page: number = data?.page || 0; + /** + * Sometimes, we're getting multiple messages being injected for the same page. + * We should lock the page to prevent same page being pulled multiple times. + */ + const lock = new Lock(this.store!, { + type: "custom", + lockKey: `clickup_pull_${job.id}_${job.config.team.id}_${job.config.folder.id}_${page}`, + ttl: 15 * 60, // 15 minutes TTL + }); + + const acquired = await lock.acquireLock("lock"); + if (!acquired) { + // Another container is processing this event, skip + throw new Error( + `Another container is processing this event ${job.id}_${job.config.team.id}_${job.config.folder.id}_${page}, skipping` + ); + } + const { last_page, tasks } = await clickUpPullService.pullTasksForFolderPaginated( job.config.team.id, job.config.folder.id, diff --git a/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts b/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts index c49308029c..5d3983b2c5 100644 --- a/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts +++ b/silo/src/apps/clickup-importer/migrator/clickup.migrator.ts @@ -221,7 +221,15 @@ export class ClickUpDataMigrator extends BaseDataMigrator= report.total_batch_count; - logger.info(`Marking job as finished for ${jobId}`, { data, jobId, phase, isLastBatch, isJobCompleted }); + logger.info(`Marking job as finished for ${jobId}`, { + data, + jobId, + phase, + isLastBatch, + isJobCompleted, + completed_batch_count: report.completed_batch_count, + total_batch_count: report.total_batch_count, + }); if (phase === E_CLICKUP_IMPORT_PHASE.ISSUES) { // check if the job is completed if (isJobCompleted) { diff --git a/silo/src/apps/engine/controllers/job.controller.ts b/silo/src/apps/engine/controllers/job.controller.ts index 56cbd8b377..f971ea289a 100644 --- a/silo/src/apps/engine/controllers/job.controller.ts +++ b/silo/src/apps/engine/controllers/job.controller.ts @@ -146,7 +146,9 @@ export class JobController { if (!job) return res.status(400).json({ message: "Job not found" }); // Dispatch an event for the importer to handle any sort of post job processing - logger.info(`[${job.id}] Dispatching job finished event for ${job.source} with phase ${phase} and isLastBatch ${isLastBatch}`); + logger.info( + `[${job.id}] Dispatching job finished event for ${job.source} with phase ${phase} and isLastBatch ${isLastBatch}` + ); await importTaskManger.registerTask( { route: job.source.toLowerCase(), @@ -201,6 +203,7 @@ export class JobController { }); return; } + logger.info(`[${job.id}] Initiating job ${job.source}}`); await client.importJob.updateImportJob(job.id, { status: "CREATED", diff --git a/silo/src/worker/base/mq.ts b/silo/src/worker/base/mq.ts index bcc1e6068c..325e0341ce 100644 --- a/silo/src/worker/base/mq.ts +++ b/silo/src/worker/base/mq.ts @@ -77,7 +77,7 @@ export class MQActorBase { private async initializeConnection() { this.connection = await amqp.connect(this.amqpUrl, { - heartbeat: 60, + heartbeat: 5 * 60, // 5 minutes - suitable for long-running operations }); this.channel = await this.connection.createConfirmChannel();