[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

@@ -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(),
         )
 

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);
           }

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,

silo/src/apps/clickup-importer/migrator/clickup.migrator.ts

@@ -221,7 +221,15 @@ export class ClickUpDataMigrator extends BaseDataMigrator<TClickUpConfig, TClick
     const job = await this.getJobData(jobId);
     const report = await integrationConnectionHelper.getImportReport({ report_id: job.report_id });
     const isJobCompleted = report.completed_batch_count >= 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) {

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",

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();