enh(zendesk): reduce batch size and add heartbeat

connectors/src/connectors/zendesk/temporal/activities.ts

@@ -18,6 +18,7 @@ import { concurrentExecutor } from "@connectors/lib/async_utils";
 import { ExternalOAuthTokenError } from "@connectors/lib/error";
 import { ZendeskTimestampCursor } from "@connectors/lib/models/zendesk";
 import { syncStarted, syncSucceeded } from "@connectors/lib/sync_status";
+import { heartbeat } from "@connectors/lib/temporal";
 import logger from "@connectors/logger/logger";
 import { ConnectorResource } from "@connectors/resources/connector_resource";
 import {
@@ -199,6 +200,7 @@ export async function syncZendeskCategoryBatchActivity({
       syncCategory({ connectorId, brandId, category, currentSyncDateMs }),
     {
       concurrency: 10,
+      onBatchComplete: heartbeat,
     }
   );
 
@@ -361,7 +363,10 @@ export async function syncZendeskArticleBatchActivity({
         loggerArgs,
         forceResync,
       }),
-    { concurrency: 10 }
+    {
+      concurrency: 10,
+      onBatchComplete: heartbeat,
+    }
   );
   return { hasMore, nextLink };
 }
@@ -446,7 +451,10 @@ export async function syncZendeskTicketBatchActivity({
         users,
       });
     },
-    { concurrency: 10 }
+    {
+      concurrency: 10,
+      onBatchComplete: heartbeat,
+    }
   );
 
   logger.info(

connectors/src/connectors/zendesk/temporal/config.ts

@@ -3,4 +3,4 @@ export const QUEUE_NAME = `zendesk-queue-v${WORKFLOW_VERSION}`;
 
 // Batch size used when fetching from Zendesk API or from the database
 // 100 is the maximum value allowed for most endpoints in Zendesk: https://developer.zendesk.com/api-reference/introduction/pagination/
-export const ZENDESK_BATCH_SIZE = 50;
+export const ZENDESK_BATCH_SIZE = 30;

connectors/src/lib/async_utils.ts

@@ -2,18 +2,33 @@ import PQueue from "p-queue";
 
 export async function concurrentExecutor<T, V>(
   items: T[],
-  iterator: (item: T, idx: number) => Promise<V>,
-  { concurrency = 8 }: { concurrency: number }
+  task: (item: T, idx: number) => Promise<V>,
+  {
+    concurrency,
+    onBatchComplete,
+  }: {
+    concurrency: number;
+    onBatchComplete?: () => Promise<void>;
+  }
 ) {
   const queue = new PQueue({ concurrency });
   const promises: Promise<V>[] = [];
 
   for (const [idx, item] of items.entries()) {
     // Queue each task. The queue manages concurrency.
     // Each task is wrapped in a promise to capture its completion.
-    const p = queue.add(async () => iterator(item, idx));
+    const p = queue.add(async () => {
+      const result = await task(item, idx);
+
+      // Call the onBatchComplete callback if it's provided and the batch is complete
+      if (onBatchComplete && (idx + 1) % concurrency === 0) {
+        await onBatchComplete();
+      }
+
+      return result;
+    });
 
-    // Cast the promise to Promise<R> to fix the return type from P-Queue.
+    // Cast the promise to Promise<V> to fix the return type from P-Queue.
     promises.push(p as Promise<V>);
   }