fix(data-drains): guard dispatcher rollback so DB errors don't abort batch

If the rollback update threw (e.g. transient DB error), the exception
bubbled out of the for loop and silently skipped the rest of the
candidate drains for the cycle. Wrap it so the batch continues.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/lib/data-drains/dispatcher.ts

@@ -1,6 +1,7 @@
 import { db } from '@sim/db'
 import { dataDrainRuns, dataDrains } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { toError } from '@sim/utils/errors'
 import { and, eq, isNull, lt, or } from 'drizzle-orm'
 import { isOrganizationOnEnterprisePlan } from '@/lib/billing/core/subscription'
 import { getJobQueue } from '@/lib/core/async-jobs'
@@ -148,11 +149,22 @@ export async function dispatchDueDrains(now: Date = new Date()): Promise<{
     } catch (error) {
       // Roll back the claim — otherwise a transient queue outage silently
       // delays this drain by a full cadence. We only revert if no other
-      // process has advanced lastRunAt past our claim timestamp.
-      await db
-        .update(dataDrains)
-        .set({ lastRunAt: candidate.lastRunAt, updatedAt: now })
-        .where(and(eq(dataDrains.id, candidate.id), eq(dataDrains.lastRunAt, now)))
+      // process has advanced lastRunAt past our claim timestamp. Guard the
+      // rollback itself so a transient DB error here doesn't abort the
+      // remaining candidates in the batch.
+      try {
+        await db
+          .update(dataDrains)
+          .set({ lastRunAt: candidate.lastRunAt, updatedAt: now })
+          .where(and(eq(dataDrains.id, candidate.id), eq(dataDrains.lastRunAt, now)))
+      } catch (rollbackError) {
+        logger.error('Failed to roll back data-drain claim after enqueue failure', {
+          drainId: candidate.id,
+          enqueueError: toError(error).message,
+          rollbackError: toError(rollbackError).message,
+        })
+        continue
+      }
       logger.error('Failed to enqueue data-drain job; rolled back claim', {
         drainId: candidate.id,
         error,