feat: nightly connector switch sweep with watermark-rewind handover

services/apps/connectors_worker/src/activities.ts

@@ -13,12 +13,14 @@ import {
   evaluateBackGate,
   evaluateForwardGate,
   isUnitLocked,
+  listSwitchSweepRepos,
   pauseNangoRepoSync,
   quiesceUnit,
-  replayShadowRecords,
+  readNangoHandoverPoint,
   resolveSwitchTarget,
   resumeNangoRepoSync,
   setUnitEmitEnabled,
+  takeOverUnit,
 } from './activities/switchSyncPipelineActivities'
 import { executeSync } from './activities/syncRunActivities'
 
@@ -32,15 +34,17 @@ export {
   executeSync,
   isUnitLocked,
   listShadowDiffChannels,
+  listSwitchSweepRepos,
   logDispatchSummary,
   pauseNangoRepoSync,
   quiesceUnit,
-  replayShadowRecords,
+  readNangoHandoverPoint,
   resolveSwitchTarget,
   resumeNangoRepoSync,
   runOctolensDiscovery,
   runShadowDiffForChannel,
   setUnitEmitEnabled,
   startRun,
+  takeOverUnit,
   touchHeartbeat,
 }

services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts

@@ -1,57 +1,40 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
-import { IntegrationResultType } from '@crowd/types'
-
 import {
   evaluateForwardGate,
   pauseNangoRepoSync,
-  replayShadowRecords,
+  readNangoHandoverPoint,
   resolveSwitchTarget,
   setUnitEmitEnabled,
+  takeOverUnit,
 } from './switchSyncPipelineActivities'
 
 const mocks = vi.hoisted(() => ({
   findSyncUnitByChannelName: vi.fn(),
   getSyncDiffCountsForDays: vi.fn(),
   setSyncUnitEmitEnabled: vi.fn(),
-  listShadowRecordsBatch: vi.fn(),
-  deleteShadowRecords: vi.fn(),
-  publishExternalResult: vi.fn(),
-  triggerResultProcessing: vi.fn(),
-  heartbeat: vi.fn(),
+  takeOverSyncUnit: vi.fn(),
   findActiveNangoMappingForRepo: vi.fn(),
   getNangoModelLastCheckedAt: vi.fn(),
   initNangoCloudClient: vi.fn(),
   pauseNangoSync: vi.fn(),
   startNangoSync: vi.fn(),
+  getNangoConnectionStatus: vi.fn(),
   log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
 }))
 
 vi.mock('../main', () => ({
   svc: {
     postgres: { writer: {} },
     log: mocks.log,
-    dataSinkWorkerEmitter: { triggerResultProcessing: mocks.triggerResultProcessing },
   },
 }))
-vi.mock('@temporalio/activity', () => ({
-  Context: { current: () => ({ heartbeat: mocks.heartbeat }) },
-}))
-vi.mock(
-  '@crowd/data-access-layer/src/old/apps/integration_stream_worker/integrationStream.repo',
-  () => ({
-    default: class {
-      publishExternalResult = mocks.publishExternalResult
-    },
-  }),
-)
 vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({ dbStoreQx: vi.fn(() => 'qx') }))
 vi.mock('@crowd/data-access-layer/src/connectors', () => ({
   findSyncUnitByChannelName: mocks.findSyncUnitByChannelName,
   getSyncDiffCountsForDays: mocks.getSyncDiffCountsForDays,
   setSyncUnitEmitEnabled: mocks.setSyncUnitEmitEnabled,
-  listShadowRecordsBatch: mocks.listShadowRecordsBatch,
-  deleteShadowRecords: mocks.deleteShadowRecords,
+  takeOverSyncUnit: mocks.takeOverSyncUnit,
 }))
 vi.mock('@crowd/data-access-layer/src/integrations', () => ({
   findActiveNangoMappingForRepo: mocks.findActiveNangoMappingForRepo,
@@ -62,6 +45,7 @@ vi.mock('@crowd/nango', async (importOriginal) => ({
   initNangoCloudClient: mocks.initNangoCloudClient,
   pauseNangoSync: mocks.pauseNangoSync,
   startNangoSync: mocks.startNangoSync,
+  getNangoConnectionStatus: mocks.getNangoConnectionStatus,
 }))
 
 const activeUnit = {
@@ -163,33 +147,35 @@ describe('switch activities', () => {
     )
   })
 
-  it('replays every remaining shadow record through the sink path, batch by batch, then deletes it', async () => {
-    const row = (sourceId: string) => ({
-      type: 'issues-opened',
-      sourceId,
-      occurredAt: '2026-10-18T00:00:00Z',
-      data: { sourceId },
-      segmentId: 'seg-1',
-      integrationId: 'int-1',
-    })
-    mocks.listShadowRecordsBatch
-      .mockResolvedValueOnce([row('a'), row('b')])
-      .mockResolvedValueOnce([row('c')])
-      .mockResolvedValueOnce([])
-    mocks.publishExternalResult.mockImplementation(
-      async (_: string, r: { data: { sourceId: string } }) => `res-${r.data.sourceId}`,
+  it('hands over from the earlier of the nango run and cdp consumption, and takes the unit over an hour before it', async () => {
+    mocks.getNangoConnectionStatus.mockResolvedValue([
+      { name: 'issues', status: 'PAUSED', finishedAt: '2026-10-19T09:00:00.000Z' },
+      { name: 'forks', status: 'SUCCESS' },
+    ])
+    mocks.getNangoModelLastCheckedAt.mockResolvedValue('2026-10-19T07:00:00Z')
+    const point = { integrationId: 'int-1', connectionId: 'conn-1' }
+    await expect(
+      readNangoHandoverPoint({ ...point, syncName: 'issues', nangoSync: 'issues' }),
+    ).resolves.toEqual({ handoverAt: '2026-10-19T07:00:00Z' })
+    expect(mocks.getNangoConnectionStatus).toHaveBeenCalledWith('github', 'conn-1')
+    expect(mocks.getNangoModelLastCheckedAt).toHaveBeenCalledWith(
+      'qx',
+      'int-1',
+      'conn-1',
+      'GithubIssue',
     )
+    await expect(
+      readNangoHandoverPoint({ ...point, syncName: 'forks', nangoSync: 'forks' }),
+    ).resolves.toEqual({ handoverAt: null })
 
-    await expect(replayShadowRecords({ unitId: 'u-1' })).resolves.toBe(3)
-
-    expect(mocks.publishExternalResult).toHaveBeenCalledTimes(3)
-    expect(mocks.publishExternalResult).toHaveBeenCalledWith('int-1', {
-      type: IntegrationResultType.ACTIVITY,
-      segmentId: 'seg-1',
-      data: { sourceId: 'c' },
+    mocks.takeOverSyncUnit.mockResolvedValue(1)
+    await expect(
+      takeOverUnit({ unitId: 'u-1', handoverAt: '2026-10-19T07:00:00Z' }),
+    ).resolves.toEqual({ since: '2026-10-19T06:00:00.000Z' })
+    expect(mocks.takeOverSyncUnit).toHaveBeenCalledWith('qx', 'u-1', '2026-10-19T06:00:00.000Z')
+    await expect(takeOverUnit({ unitId: 'u-2', handoverAt: null })).resolves.toEqual({
+      since: null,
     })
-    expect(mocks.triggerResultProcessing).toHaveBeenCalledWith('res-c', 'res-c', false)
-    expect(mocks.deleteShadowRecords.mock.calls.map(([, , batch]) => batch.length)).toEqual([2, 1])
-    expect(mocks.heartbeat).toHaveBeenLastCalledWith(3)
+    expect(mocks.takeOverSyncUnit).toHaveBeenLastCalledWith('qx', 'u-2', null)
   })
 })

services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.ts

@@ -1,45 +1,44 @@
-import { Context } from '@temporalio/activity'
-
 import {
   activateSyncUnit,
-  deleteShadowRecords,
   findSyncUnitByChannelName,
   getSyncDiffCountsForDays,
   isSyncUnitLocked,
-  listShadowRecordsBatch,
+  listShadowDiffUnits,
   pauseSyncUnit,
   setSyncUnitEmitEnabled,
+  takeOverSyncUnit,
 } from '@crowd/data-access-layer/src/connectors'
 import {
   findActiveNangoMappingForRepo,
   getNangoModelLastCheckedAt,
 } from '@crowd/data-access-layer/src/integrations'
-import IntegrationStreamRepository from '@crowd/data-access-layer/src/old/apps/integration_stream_worker/integrationStream.repo'
 import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor'
 import {
   NangoIntegration,
   getGithubSyncForConnectorSync,
+  getNangoConnectionStatus,
   githubChannelName,
   initNangoCloudClient,
   pauseNangoSync,
   startNangoSync,
 } from '@crowd/nango'
-import { IntegrationResultType } from '@crowd/types'
 
 import { svc } from '../main'
 import {
   evaluateBackUnitGate,
   evaluateDiffCleanliness,
   evaluateForwardUnitGate,
   evaluateNangoConsumption,
+  groupUnitsForSweep,
+  handoverPoint,
+  handoverSince,
   recentDiffedDays,
   resolveSwitchTargets,
 } from '../syncPipelineSwitch'
-import type { IGateRefusal, ISwitchSyncTarget } from '../syncPipelineSwitch'
+import type { IGateRefusal, ISweepRepo, ISwitchSyncTarget } from '../syncPipelineSwitch'
 
 const PLATFORM = 'github'
 const SWITCH_EVENT = 'sync_pipeline_switch'
-const REPLAY_BATCH_SIZE = 200
 
 export interface ISwitchRepoTarget {
   integrationId: string
@@ -81,6 +80,26 @@ export interface IUnitInput {
   unitId: string
 }
 
+export interface IHandoverPointInput {
+  integrationId: string
+  connectionId: string
+  syncName: string
+  nangoSync: string
+}
+
+export interface IHandoverPoint {
+  handoverAt: string | null
+}
+
+export interface ITakeOverUnitInput {
+  unitId: string
+  handoverAt: string | null
+}
+
+export interface ITakeOverUnitResult {
+  since: string | null
+}
+
 export async function resolveSwitchTarget(
   input: IResolveSwitchTargetInput,
 ): Promise<IResolveSwitchTargetResult> {
@@ -191,31 +210,48 @@ export async function activateUnit(input: IUnitInput): Promise<void> {
   svc.log.info({ event: SWITCH_EVENT, ...input, activated: activated === 1 }, 'sync unit activated')
 }
 
-export async function replayShadowRecords(input: IUnitInput): Promise<number> {
-  const qx = dbStoreQx(svc.postgres.writer)
-  const streamRepo = new IntegrationStreamRepository(svc.postgres.writer, svc.log)
-  let replayed = 0
-
-  for (;;) {
-    const batch = await listShadowRecordsBatch(qx, input.unitId, REPLAY_BATCH_SIZE)
-    if (batch.length === 0) {
-      break
-    }
-
-    for (const record of batch) {
-      const resultId = await streamRepo.publishExternalResult(record.integrationId, {
-        type: IntegrationResultType.ACTIVITY,
-        segmentId: record.segmentId,
-        data: record.data,
-      })
-      await svc.dataSinkWorkerEmitter.triggerResultProcessing(resultId, resultId, false)
-    }
-
-    await deleteShadowRecords(qx, input.unitId, batch)
-    replayed += batch.length
-    Context.current().heartbeat(replayed)
-  }
+export async function readNangoHandoverPoint(input: IHandoverPointInput): Promise<IHandoverPoint> {
+  await initNangoCloudClient()
+  const syncs = await getNangoConnectionStatus(NangoIntegration.GITHUB, input.connectionId)
+  const sync = syncs?.find((s) => s.name === input.nangoSync)
+  const nangoRunFinishedAt = sync?.finishedAt ?? null
+  const consumedAt = await getNangoModelLastCheckedAt(
+    dbStoreQx(svc.postgres.writer),
+    input.integrationId,
+    input.connectionId,
+    getGithubSyncForConnectorSync(input.syncName).model,
+  )
+  const handoverAt = handoverPoint(nangoRunFinishedAt, consumedAt)
+  svc.log.info(
+    {
+      event: SWITCH_EVENT,
+      ...input,
+      nangoStatus: sync?.status ?? null,
+      nangoRunFinishedAt,
+      consumedAt,
+      handoverAt,
+    },
+    'nango handover point read',
+  )
+  return { handoverAt }
+}
 
-  svc.log.info({ event: SWITCH_EVENT, ...input, replayed }, 'shadow records replayed')
-  return replayed
+export async function takeOverUnit(input: ITakeOverUnitInput): Promise<ITakeOverUnitResult> {
+  const since = input.handoverAt ? handoverSince(input.handoverAt) : null
+  const flipped = await takeOverSyncUnit(dbStoreQx(svc.postgres.writer), input.unitId, since)
+  svc.log.info(
+    { event: SWITCH_EVENT, ...input, since, alreadyEmitting: flipped === 0 },
+    'sync unit taken over',
+  )
+  return { since }
+}
+
+export async function listSwitchSweepRepos(): Promise<ISweepRepo[]> {
+  const units = await listShadowDiffUnits(dbStoreQx(svc.postgres.writer))
+  const repos = groupUnitsForSweep(units)
+  svc.log.info(
+    { event: SWITCH_EVENT, repos: repos.length, units: units.length },
+    'sweep candidates listed',
+  )
+  return repos
 }

services/apps/connectors_worker/src/bin/switch-sync-pipeline.ts

@@ -2,6 +2,7 @@ import { getServiceLogger } from '@crowd/logging'
 import { TEMPORAL_CONFIG, getTemporalClient } from '@crowd/temporal'
 
 import type { ISwitchSyncResult, SwitchDirection } from '../syncPipelineSwitch'
+import { switchSyncPipelineWorkflowId } from '../workflows/switchSyncPipeline'
 import { parseRepoArg } from './repoArg'
 
 const log = getServiceLogger()
@@ -45,7 +46,7 @@ setImmediate(async () => {
   try {
     const input = parseArgs(process.argv.slice(2))
     const temporal = await getTemporalClient(TEMPORAL_CONFIG())
-    const workflowId = `connectors-switch-sync-pipeline/${input.owner.toLowerCase()}/${input.repo.toLowerCase()}`
+    const workflowId = switchSyncPipelineWorkflowId(input.owner, input.repo)
 
     log.info({ workflowId, ...input }, 'starting sync pipeline switch')
     const handle = await temporal.workflow.start('switchSyncPipeline', {

services/apps/connectors_worker/src/main.ts

@@ -9,6 +9,7 @@ import { dummyConnector } from '@crowd/connectors/src/testing/dummyConnector'
 import { scheduleDispatcher } from './schedules/dispatcher'
 import { scheduleOctolensDiscovery } from './schedules/octolensDiscovery'
 import { scheduleShadowDiff } from './schedules/shadowDiff'
+import { scheduleSwitchSyncPipelineSweep } from './schedules/switchSyncPipelineSweep'
 
 const config: Config = {
   envvars: [],
@@ -48,6 +49,7 @@ setImmediate(async () => {
   await scheduleDispatcher()
   await scheduleShadowDiff()
   await scheduleOctolensDiscovery()
+  await scheduleSwitchSyncPipelineSweep()
 
   await svc.start()
 })

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

@@ -0,0 +1,31 @@
+import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
+
+import { svc } from '../main'
+import { switchSyncPipelineSweep } from '../workflows/switchSyncPipelineSweep'
+
+export async function scheduleSwitchSyncPipelineSweep(): Promise<void> {
+  try {
+    await svc.temporal.schedule.create({
+      scheduleId: 'connectors-switch-sync-pipeline-sweep',
+      spec: {
+        calendars: [{ hour: 2, minute: 0 }],
+      },
+      policies: {
+        overlap: ScheduleOverlapPolicy.SKIP,
+        catchupWindow: '1 hour',
+      },
+      action: {
+        type: 'startWorkflow',
+        workflowType: switchSyncPipelineSweep,
+        taskQueue: 'connectors',
+        args: [],
+      },
+    })
+  } catch (err) {
+    if (err instanceof ScheduleAlreadyRunning) {
+      svc.log.info('Switch sync pipeline sweep schedule already registered in Temporal.')
+    } else {
+      throw new Error(err)
+    }
+  }
+}