fix(polling): retry failed idempotency keys, fix drive cursor overshoot, fix calendar inclusive updatedMin

Author: waleedlatif1

apps/sim/lib/core/idempotency/service.ts

@@ -13,6 +13,8 @@ const logger = createLogger('IdempotencyService')
 export interface IdempotencyConfig {
   ttlSeconds?: number
   namespace?: string
+  /** When true, failed keys are deleted rather than stored so the operation is retried on the next attempt. */
+  retryFailures?: boolean
 }
 
 export interface IdempotencyResult {
@@ -58,6 +60,7 @@ export class IdempotencyService {
     this.config = {
       ttlSeconds: config.ttlSeconds ?? DEFAULT_TTL,
       namespace: config.namespace ?? 'default',
+      retryFailures: config.retryFailures ?? false,
     }
     this.storageMethod = getStorageMethod()
     logger.info(`IdempotencyService using ${this.storageMethod} storage`, {
@@ -340,6 +343,21 @@ export class IdempotencyService {
     logger.debug(`Stored idempotency result in database: ${normalizedKey}`)
   }
 
+  private async deleteKey(
+    normalizedKey: string,
+    storageMethod: 'redis' | 'database'
+  ): Promise<void> {
+    if (storageMethod === 'redis') {
+      const redis = getRedisClient()
+      if (redis) await redis.del(`${REDIS_KEY_PREFIX}${normalizedKey}`).catch(() => {})
+    } else {
+      await db
+        .delete(idempotencyKey)
+        .where(eq(idempotencyKey.key, normalizedKey))
+        .catch(() => {})
+    }
+  }
+
   async executeWithIdempotency<T>(
     provider: string,
     identifier: string,
@@ -360,6 +378,10 @@ export class IdempotencyService {
       }
 
       if (existingResult?.status === 'failed') {
+        if (this.config.retryFailures) {
+          await this.deleteKey(claimResult.normalizedKey, claimResult.storageMethod)
+          return this.executeWithIdempotency(provider, identifier, operation, additionalContext)
+        }
         logger.info(`Previous operation failed for: ${claimResult.normalizedKey}`)
         throw new Error(existingResult.error || 'Previous operation failed')
       }
@@ -391,11 +413,15 @@ export class IdempotencyService {
     } catch (error) {
       const errorMessage = error instanceof Error ? error.message : 'Unknown error'
 
-      await this.storeResult(
-        claimResult.normalizedKey,
-        { success: false, error: errorMessage, status: 'failed' },
-        claimResult.storageMethod
-      )
+      if (this.config.retryFailures) {
+        await this.deleteKey(claimResult.normalizedKey, claimResult.storageMethod)
+      } else {
+        await this.storeResult(
+          claimResult.normalizedKey,
+          { success: false, error: errorMessage, status: 'failed' },
+          claimResult.storageMethod
+        )
+      }
 
       logger.warn(`Operation failed: ${claimResult.normalizedKey} - ${errorMessage}`)
       throw error
@@ -454,4 +480,5 @@ export const webhookIdempotency = new IdempotencyService({
 export const pollingIdempotency = new IdempotencyService({
   namespace: 'polling',
   ttlSeconds: 60 * 60 * 24 * 3, // 3 days
+  retryFailures: true,
 })

apps/sim/lib/webhooks/polling/google-calendar.ts

@@ -143,7 +143,9 @@ export const googleCalendarPollingHandler: PollingProviderHandler = {
       const newTimestamp =
         processedCount === 0 && failedCount > 0
           ? config.lastCheckedTimestamp
-          : latestUpdated || now.toISOString()
+          : latestUpdated
+            ? new Date(new Date(latestUpdated).getTime() + 1).toISOString()
+            : now.toISOString()
       await updateWebhookProviderConfig(webhookId, { lastCheckedTimestamp: newTimestamp }, logger)
 
       if (failedCount > 0 && processedCount === 0) {

apps/sim/lib/webhooks/polling/google-drive.ts

@@ -255,24 +255,12 @@ async function fetchChanges(
       newStartPageToken = data.newStartPageToken as string
     }
 
-    // Only advance the resume cursor when we'll actually use all changes from this page.
-    // If allChanges exceeds maxFiles, we'll slice off the extras — so we must NOT
-    // advance past this page, otherwise the sliced changes are lost permanently.
     const hasMore = !!data.nextPageToken
     const overLimit = allChanges.length >= maxFiles
 
     if (!hasMore || overLimit || pages >= MAX_PAGES) {
-      // If we stopped mid-stream and haven't consumed all changes from this page,
-      // keep currentPageToken so the next poll re-fetches this page.
-      // If we consumed everything on this page but there are more pages,
-      // advance to nextPageToken so we don't re-process this page.
       if (hasMore && !overLimit) {
         lastNextPageToken = data.nextPageToken as string
-      } else if (hasMore && overLimit && allChanges.length > maxFiles) {
-        // We got more changes than maxFiles from this page — don't advance,
-        // re-fetch this page next time (idempotency deduplicates already-processed ones)
-      } else if (hasMore) {
-        lastNextPageToken = data.nextPageToken as string
       }
       break
     }
@@ -281,7 +269,10 @@ async function fetchChanges(
     currentPageToken = data.nextPageToken as string
   }
 
-  const resumeToken = newStartPageToken ?? lastNextPageToken ?? config.pageToken!
+  const slicingOccurs = allChanges.length > maxFiles
+  const resumeToken = slicingOccurs
+    ? (lastNextPageToken ?? config.pageToken!)
+    : (newStartPageToken ?? lastNextPageToken ?? config.pageToken!)
 
   return { changes: allChanges.slice(0, maxFiles), newStartPageToken: resumeToken }
 }