feat: diff 50 shadow channels at a time with a four-hour run budget

services/apps/connectors_worker/src/schedules/shadowDiff.ts

@@ -3,28 +3,33 @@ import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/clien
 import { svc } from '../main'
 import { shadowDiff } from '../workflows/shadowDiff'
 
+const SCHEDULE_ID = 'connectors-shadow-diff'
+
+const action = {
+  type: 'startWorkflow',
+  workflowType: shadowDiff,
+  taskQueue: 'connectors',
+  args: [],
+  workflowExecutionTimeout: '4 hours',
+} as const
+
 export async function scheduleShadowDiff(): Promise<void> {
   try {
     await svc.temporal.schedule.create({
-      scheduleId: 'connectors-shadow-diff',
+      scheduleId: SCHEDULE_ID,
       spec: {
         calendars: [{ hour: { start: 0, end: 0 } }],
       },
       policies: {
         overlap: ScheduleOverlapPolicy.SKIP,
         catchupWindow: '1 hour',
       },
-      action: {
-        type: 'startWorkflow',
-        workflowType: shadowDiff,
-        taskQueue: 'connectors',
-        args: [],
-        workflowExecutionTimeout: '30 minutes',
-      },
+      action,
     })
   } catch (err) {
     if (err instanceof ScheduleAlreadyRunning) {
-      svc.log.info('Shadow diff schedule already registered in Temporal.')
+      await svc.temporal.schedule.getHandle(SCHEDULE_ID).update((prev) => ({ ...prev, action }))
+      svc.log.info('Shadow diff schedule already registered in Temporal, action reconciled.')
     } else {
       throw new Error(err)
     }

services/apps/connectors_worker/src/workflows/shadowDiff.test.ts

@@ -1,6 +1,20 @@
-import { describe, expect, it } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-import { describeChannelError } from './shadowDiff'
+import { CHANNEL_CONCURRENCY, describeChannelError, shadowDiff } from './shadowDiff'
+
+const mocks = vi.hoisted(() => ({
+  listShadowDiffChannels: vi.fn(),
+  runShadowDiffForChannel: vi.fn(),
+  log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+}))
+
+vi.mock('@temporalio/workflow', () => ({
+  proxyActivities: () => ({
+    listShadowDiffChannels: mocks.listShadowDiffChannels,
+    runShadowDiffForChannel: mocks.runShadowDiffForChannel,
+  }),
+  log: mocks.log,
+}))
 
 describe('describeChannelError', () => {
   it('extracts the message from an Error so operators can see why a channel failed', () => {
@@ -18,3 +32,35 @@ describe('describeChannelError', () => {
     expect(describeChannelError(circular)).toBe(String(circular))
   })
 })
+
+describe('shadowDiff', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('diffs channels 50 at a time and keeps counting after a failed channel', async () => {
+    const channels = Array.from({ length: 120 }, (_, i) => ({
+      channelName: `c${i}`,
+      integrationId: 'i',
+      units: [],
+    }))
+    mocks.listShadowDiffChannels.mockResolvedValue(channels)
+    let inFlight = 0
+    let peak = 0
+    mocks.runShadowDiffForChannel.mockImplementation(async (channel) => {
+      inFlight += 1
+      peak = Math.max(peak, inFlight)
+      await Promise.resolve()
+      inFlight -= 1
+      if (channel.channelName === 'c7') throw new Error('boom')
+      return { channelName: channel.channelName, integrationId: 'i', status: 'ok' }
+    })
+    await shadowDiff()
+    expect(peak).toBe(CHANNEL_CONCURRENCY)
+    expect(CHANNEL_CONCURRENCY).toBe(50)
+    expect(mocks.log.info).toHaveBeenCalledWith(
+      'shadow diff run complete',
+      expect.objectContaining({ channelCount: 120, okCount: 119, errorCount: 1 }),
+    )
+  })
+})

services/apps/connectors_worker/src/workflows/shadowDiff.ts

@@ -7,7 +7,7 @@ const activity = proxyActivities<typeof activities>({
   retry: { maximumAttempts: 3, backoffCoefficient: 2 },
 })
 
-const CHANNEL_CONCURRENCY = 10
+export const CHANNEL_CONCURRENCY = 50
 
 export function describeChannelError(err: unknown): string {
   if (err instanceof Error) {