diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts index de078c9e5e..fad4b0612b 100644 --- a/services/apps/connectors_worker/src/activities.ts +++ b/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, } diff --git a/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts b/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts index e896a25fc0..c404bb7ffa 100644 --- a/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts @@ -1,29 +1,25 @@ 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() }, })) @@ -31,27 +27,14 @@ 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) }) }) diff --git a/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.ts b/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.ts index 1d4ef0644f..3181074654 100644 --- a/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.ts +++ b/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.ts @@ -1,30 +1,27 @@ -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 { @@ -32,14 +29,16 @@ import { 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 { @@ -191,31 +210,48 @@ export async function activateUnit(input: IUnitInput): Promise { svc.log.info({ event: SWITCH_EVENT, ...input, activated: activated === 1 }, 'sync unit activated') } -export async function replayShadowRecords(input: IUnitInput): Promise { - 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 { + 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 { + 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 { + 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 } diff --git a/services/apps/connectors_worker/src/bin/switch-sync-pipeline.ts b/services/apps/connectors_worker/src/bin/switch-sync-pipeline.ts index 719fd37482..322d66239a 100644 --- a/services/apps/connectors_worker/src/bin/switch-sync-pipeline.ts +++ b/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', { diff --git a/services/apps/connectors_worker/src/main.ts b/services/apps/connectors_worker/src/main.ts index 4bb3949a3c..2cbfae1ff7 100644 --- a/services/apps/connectors_worker/src/main.ts +++ b/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() }) diff --git a/services/apps/connectors_worker/src/schedules/switchSyncPipelineSweep.ts b/services/apps/connectors_worker/src/schedules/switchSyncPipelineSweep.ts new file mode 100644 index 0000000000..db0ec14281 --- /dev/null +++ b/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 { + 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) + } + } +} diff --git a/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts b/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts index 0570912392..21d53e66a8 100644 --- a/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts +++ b/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts @@ -8,6 +8,9 @@ import { evaluateDiffCleanliness, evaluateForwardUnitGate, evaluateNangoConsumption, + groupUnitsForSweep, + handoverPoint, + handoverSince, recentDiffedDays, resolveSwitchTargets, } from './syncPipelineSwitch' @@ -109,3 +112,39 @@ describe('back gate', () => { expect(evaluateForwardUnitGate(unit({ emitEnabled: true }))).toMatchObject({ unitId: 'u-1' }) }) }) + +describe('groupUnitsForSweep', () => { + it('turns shadow units into one repo per channel with only its unswitched syncs', () => { + const shadowUnit = (channelName: string, syncName: string, integrationId = 'int-1') => ({ + id: `${channelName}/${syncName}`, + integrationId, + channelName, + syncName, + }) + + expect( + groupUnitsForSweep([ + shadowUnit('https://github.com/Org/Repo', 'issues'), + shadowUnit('https://github.com/other/thing', 'forks'), + shadowUnit('https://github.com/org/repo', 'pull-requests', 'int-2'), + ]), + ).toEqual([ + { owner: 'Org', repo: 'Repo', syncNames: ['issues', 'pull-requests'] }, + { owner: 'other', repo: 'thing', syncNames: ['forks'] }, + ]) + }) +}) + +describe('handover point', () => { + it('hands over from the earlier of the nango run finish and cdp consumption, an hour back', () => { + expect(handoverPoint('2026-10-19T11:00:00Z', '2026-10-19T08:00:00Z')).toBe( + '2026-10-19T08:00:00Z', + ) + expect(handoverPoint('2026-10-19T09:00:00Z', '2026-10-19T09:30:00Z')).toBe( + '2026-10-19T09:00:00Z', + ) + expect(handoverPoint(null, '2026-10-19T09:30:00Z')).toBeNull() + expect(handoverPoint('2026-10-19T09:00:00Z', null)).toBeNull() + expect(handoverSince('2026-10-19T09:00:00Z')).toBe('2026-10-19T08:00:00.000Z') + }) +}) diff --git a/services/apps/connectors_worker/src/syncPipelineSwitch.ts b/services/apps/connectors_worker/src/syncPipelineSwitch.ts index b3ce643339..65b8aa44f1 100644 --- a/services/apps/connectors_worker/src/syncPipelineSwitch.ts +++ b/services/apps/connectors_worker/src/syncPipelineSwitch.ts @@ -1,4 +1,9 @@ -import type { ISyncDiffDayCounts, ISyncUnit } from '@crowd/data-access-layer/src/connectors' +import { parseRepoChannel } from '@crowd/connectors/src/connectors/github/paging' +import type { + IShadowDiffUnit, + ISyncDiffDayCounts, + ISyncUnit, +} from '@crowd/data-access-layer/src/connectors' import { GITHUB_SYNC_MAPPINGS, getGithubSyncForConnectorSync } from '@crowd/nango' export type SwitchDirection = 'connectors' | 'nango' @@ -24,10 +29,17 @@ export interface IGateRefusal { export type GateVerdict = { ok: true } | IGateRefusal +export interface ISweepRepo { + owner: string + repo: string + syncNames: string[] +} + const MS_PER_DAY = 24 * 60 * 60 * 1000 const DIFF_LAG_DAYS = 2 export const CLEAN_DAYS_REQUIRED = 3 export const NANGO_CONSUMPTION_MAX_AGE_MS = 24 * 60 * 60 * 1000 +export const NANGO_HANDOVER_OVERLAP_MS = 60 * 60 * 1000 export function resolveSwitchTargets(requested?: string[]): ISwitchSyncTarget[] { const syncNames = requested?.length ? requested : GITHUB_SYNC_MAPPINGS.map((m) => m.connectorSync) @@ -124,3 +136,31 @@ export function evaluateBackUnitGate(unit: ISyncUnit | null): GateVerdict { } return { ok: true } } + +export function groupUnitsForSweep(units: IShadowDiffUnit[]): ISweepRepo[] { + const byRepo = new Map() + for (const unit of units) { + const key = unit.channelName.toLowerCase() + const existing = byRepo.get(key) + if (existing) { + existing.syncNames.push(unit.syncName) + } else { + byRepo.set(key, { ...parseRepoChannel(unit.channelName), syncNames: [unit.syncName] }) + } + } + return [...byRepo.values()] +} + +export function handoverPoint( + nangoRunFinishedAt: string | null, + consumedAt: string | null, +): string | null { + if (!nangoRunFinishedAt || !consumedAt) { + return null + } + return new Date(nangoRunFinishedAt) < new Date(consumedAt) ? nangoRunFinishedAt : consumedAt +} + +export function handoverSince(handoverAt: string): string { + return new Date(new Date(handoverAt).getTime() - NANGO_HANDOVER_OVERLAP_MS).toISOString() +} diff --git a/services/apps/connectors_worker/src/workflows.ts b/services/apps/connectors_worker/src/workflows.ts index 08abc54cd1..98ccc54b03 100644 --- a/services/apps/connectors_worker/src/workflows.ts +++ b/services/apps/connectors_worker/src/workflows.ts @@ -2,6 +2,14 @@ import { dispatcher } from './workflows/dispatcher' import { octolensDiscovery } from './workflows/octolensDiscovery' import { shadowDiff } from './workflows/shadowDiff' import { switchSyncPipeline } from './workflows/switchSyncPipeline' +import { switchSyncPipelineSweep } from './workflows/switchSyncPipelineSweep' import { syncRun } from './workflows/syncRun' -export { dispatcher, octolensDiscovery, shadowDiff, switchSyncPipeline, syncRun } +export { + dispatcher, + octolensDiscovery, + shadowDiff, + switchSyncPipeline, + switchSyncPipelineSweep, + syncRun, +} diff --git a/services/apps/connectors_worker/src/workflows/switchSyncPipeline.test.ts b/services/apps/connectors_worker/src/workflows/switchSyncPipeline.test.ts index 9ca72c538b..398f297a76 100644 --- a/services/apps/connectors_worker/src/workflows/switchSyncPipeline.test.ts +++ b/services/apps/connectors_worker/src/workflows/switchSyncPipeline.test.ts @@ -12,7 +12,8 @@ const activities = vi.hoisted(() => ({ pauseNangoRepoSync: vi.fn(), resumeNangoRepoSync: vi.fn(), setUnitEmitEnabled: vi.fn(), - replayShadowRecords: vi.fn(), + readNangoHandoverPoint: vi.fn(), + takeOverUnit: vi.fn(), })) vi.mock('@temporalio/workflow', () => ({ @@ -51,10 +52,11 @@ describe('switchSyncPipeline', () => { activities.pauseNangoRepoSync.mockResolvedValue(undefined) activities.resumeNangoRepoSync.mockResolvedValue(undefined) activities.setUnitEmitEnabled.mockResolvedValue(undefined) - activities.replayShadowRecords.mockResolvedValue(3) + activities.readNangoHandoverPoint.mockResolvedValue({ handoverAt: '2026-10-19T09:00:00Z' }) + activities.takeOverUnit.mockResolvedValue({ since: '2026-10-19T08:00:00.000Z' }) }) - it('forward: quiesces, drains on the lock, re-gates, pauses nango, flips, replays, reactivates; a failed pause reactivates and never flips', async () => { + it('forward: quiesces, drains on the lock, re-gates, reads the handover point, pauses nango, takes the unit over, reactivates; a failed pause reactivates and never flips', async () => { activities.evaluateForwardGate.mockImplementation(async ({ syncName }) => ok(syncName)) activities.isUnitLocked.mockResolvedValueOnce(true).mockResolvedValue(false) activities.pauseNangoRepoSync @@ -66,25 +68,35 @@ describe('switchSyncPipeline', () => { const results = await switchSyncPipeline({ owner: 'o', repo: 'r', direction: 'connectors' }) expect(results).toEqual([ - { syncName: 'issues', outcome: 'switched', reason: '3 shadow records replayed' }, + { + syncName: 'issues', + outcome: 'switched', + reason: 'since rewound to 2026-10-19T08:00:00.000Z', + }, { syncName: 'forks', outcome: 'refused', reason: 'nango pause failed: nango 503' }, ]) const order = [ activities.quiesceUnit, activities.isUnitLocked, + activities.readNangoHandoverPoint, activities.pauseNangoRepoSync, - activities.setUnitEmitEnabled, - activities.replayShadowRecords, + activities.takeOverUnit, activities.activateUnit, ].map(firstCall) expect(order).toEqual([...order].sort((a, b) => a - b)) expect(activities.evaluateForwardGate).toHaveBeenCalledTimes(4) expect(activities.evaluateForwardGate).toHaveBeenCalledWith({ ...target, syncName: 'issues' }) expect(activities.isUnitLocked).toHaveBeenCalledTimes(3) - expect(activities.setUnitEmitEnabled).toHaveBeenCalledTimes(1) - expect(activities.setUnitEmitEnabled).toHaveBeenCalledWith({ + expect(activities.setUnitEmitEnabled).not.toHaveBeenCalled() + expect(activities.takeOverUnit).toHaveBeenCalledTimes(1) + expect(activities.readNangoHandoverPoint).toHaveBeenCalledWith({ + ...target, + syncName: 'issues', + nangoSync: 'issues', + }) + expect(activities.takeOverUnit).toHaveBeenCalledWith({ unitId: 'u-issues', - emitEnabled: true, + handoverAt: '2026-10-19T09:00:00Z', }) expect(activities.activateUnit.mock.calls).toEqual([ [{ unitId: 'u-issues' }], @@ -92,7 +104,7 @@ describe('switchSyncPipeline', () => { ]) }) - it('forward: refusals, a stuck lock, a unit that changed while draining, and an emitting unit that gets healed', async () => { + it('forward: refusals, a stuck lock, a unit that changed while draining, a nango sync that never ran, a status read that keeps failing, a takeover that fails after the pause, and an emitting unit that gets healed', async () => { activities.resolveSwitchTarget.mockResolvedValue({ target, syncs: [ @@ -100,9 +112,20 @@ describe('switchSyncPipeline', () => { { syncName: 'stars', nangoSync: null }, sync('forks'), sync('pull-requests'), + sync('issue-comments'), + sync('pull-request-comments'), + sync('pull-request-commits'), sync('discussions'), ], }) + activities.readNangoHandoverPoint.mockImplementation(async ({ nangoSync }) => { + if (nangoSync === 'pull-request-comments') throw new Error('nango 502') + return { handoverAt: nangoSync === 'issue-comments' ? null : '2026-10-19T09:00:00Z' } + }) + activities.takeOverUnit.mockImplementation(async ({ unitId }) => { + if (unitId === 'u-pull-request-commits') throw new Error('deadlock') + return { since: null } + }) const seen: Record = {} activities.evaluateForwardGate.mockImplementation(async ({ syncName }) => { seen[syncName] = (seen[syncName] ?? 0) + 1 @@ -131,22 +154,47 @@ describe('switchSyncPipeline', () => { reason: "unit changed while draining: unit status is 'dead_letter' (expected active or paused)", }, + { + syncName: 'issue-comments', + outcome: 'refused', + reason: 'nango sync has no completed run to hand over from', + }, + { + syncName: 'pull-request-comments', + outcome: 'refused', + reason: 'nango status read failed: nango 502', + }, + { + syncName: 'pull-request-commits', + outcome: 'refused', + reason: 'unit takeover failed: deadlock; nango resumed; unit back in shadow', + }, { syncName: 'discussions', outcome: 'unchanged', - reason: 'already emitting; 3 shadow records replayed', + reason: 'already emitting; shadow records cleared', }, ]) - expect(activities.pauseNangoRepoSync).toHaveBeenCalledTimes(1) - expect(activities.pauseNangoRepoSync).toHaveBeenCalledWith({ + expect(activities.pauseNangoRepoSync.mock.calls.map(([i]) => i.nangoSync)).toEqual([ + 'pull-request-commits', + 'discussions', + ]) + expect(activities.resumeNangoRepoSync).toHaveBeenCalledWith({ connectionId: 'conn-1', - nangoSync: 'discussions', + nangoSync: 'pull-request-commits', }) expect(activities.setUnitEmitEnabled).not.toHaveBeenCalled() - expect(activities.replayShadowRecords).toHaveBeenCalledWith({ unitId: 'u-discussions' }) + expect(activities.takeOverUnit).toHaveBeenCalledTimes(2) + expect(activities.takeOverUnit).toHaveBeenLastCalledWith({ + unitId: 'u-discussions', + handoverAt: null, + }) expect(activities.activateUnit.mock.calls).toEqual([ [{ unitId: 'u-forks' }], [{ unitId: 'u-pull-requests' }], + [{ unitId: 'u-issue-comments' }], + [{ unitId: 'u-pull-request-comments' }], + [{ unitId: 'u-pull-request-commits' }], [{ unitId: 'u-discussions' }], ]) }) diff --git a/services/apps/connectors_worker/src/workflows/switchSyncPipeline.ts b/services/apps/connectors_worker/src/workflows/switchSyncPipeline.ts index 3f37920eeb..615969b1b4 100644 --- a/services/apps/connectors_worker/src/workflows/switchSyncPipeline.ts +++ b/services/apps/connectors_worker/src/workflows/switchSyncPipeline.ts @@ -12,9 +12,8 @@ const activity = proxyActivities({ }, }) -const replayActivity = proxyActivities>({ - startToCloseTimeout: '1 hour', - heartbeatTimeout: '2 minutes', +const takeOverActivity = proxyActivities>({ + startToCloseTimeout: '10 minutes', retry: { maximumAttempts: 3, backoffCoefficient: 2, @@ -24,6 +23,10 @@ const replayActivity = proxyActivities { + let nangoState = 'nango resumed' try { - await activity.pauseNangoRepoSync({ + await activity.resumeNangoRepoSync({ connectionId: target.connectionId, nangoSync: sync.nangoSync, }) } catch (err) { - return refused(sync, `nango pause failed: ${describeFailure(err)}`) + nangoState = `nango still paused, resume failed: ${describeFailure(err)}` } - - return finishTakeover(sync, unitId, 'unchanged', reason) + await activity.activateUnit({ unitId }) + return refused(sync, `${reason}; ${nangoState}; unit back in shadow`) } -async function finishTakeover( +async function healEmittingUnit( + target: ISwitchRepoTarget, sync: ISwitchSyncTarget, unitId: string, - outcome: 'switched' | 'unchanged', - reason?: string, + reason: string, ): Promise { - let replayed: number try { - replayed = await replayActivity.replayShadowRecords({ unitId }) + await activity.pauseNangoRepoSync({ + connectionId: target.connectionId, + nangoSync: sync.nangoSync, + }) } catch (err) { - return refused( - sync, - `emitting but shadow replay failed: ${describeFailure(err)}; rerun to finish`, - ) + return refused(sync, `nango pause failed: ${describeFailure(err)}`) } - await activity.activateUnit({ unitId }) + try { + await takeOverActivity.takeOverUnit({ unitId, handoverAt: null }) + } catch (err) { + return refused(sync, `emitting but shadow cleanup failed: ${describeFailure(err)}; rerun`) + } - const detail = `${replayed} shadow records replayed` - return { syncName: sync.syncName, outcome, reason: reason ? `${reason}; ${detail}` : detail } + await activity.activateUnit({ unitId }) + return { + syncName: sync.syncName, + outcome: 'unchanged', + reason: `${reason}; shadow records cleared`, + } } async function drainUnit(unitId: string): Promise { diff --git a/services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.test.ts b/services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.test.ts new file mode 100644 index 0000000000..12cea66664 --- /dev/null +++ b/services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.test.ts @@ -0,0 +1,74 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { switchSyncPipelineSweep } from './switchSyncPipelineSweep' + +const mocks = vi.hoisted(() => ({ + listSwitchSweepRepos: vi.fn(), + executeChild: vi.fn(), + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})) + +vi.mock('@temporalio/workflow', () => ({ + proxyActivities: () => ({ listSwitchSweepRepos: mocks.listSwitchSweepRepos }), + executeChild: mocks.executeChild, + ParentClosePolicy: { ABANDON: 'ABANDON' }, + log: mocks.log, +})) +vi.mock('./switchSyncPipeline', () => ({ + switchSyncPipeline: 'switchSyncPipeline', + switchSyncPipelineWorkflowId: (owner: string, repo: string) => `id/${owner}/${repo}`, +})) + +describe('switchSyncPipelineSweep', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('switches each repo forward in series with its own unswitched syncs, skips a repo whose switch is already running, and survives a failed child', async () => { + mocks.listSwitchSweepRepos.mockResolvedValue([ + { owner: 'a', repo: 'one', syncNames: ['issues', 'forks'] }, + { owner: 'b', repo: 'two', syncNames: ['issues'] }, + { owner: 'c', repo: 'three', syncNames: ['pull-requests'] }, + ]) + mocks.executeChild + .mockResolvedValueOnce([ + { syncName: 'issues', outcome: 'switched' }, + { syncName: 'forks', outcome: 'refused', reason: 'no diff summary for 2026-10-16' }, + ]) + .mockRejectedValueOnce( + Object.assign(new Error('running'), { name: 'WorkflowExecutionAlreadyStartedError' }), + ) + .mockRejectedValueOnce(new Error('child timed out')) + + await expect(switchSyncPipelineSweep()).resolves.toEqual({ + repos: 3, + switched: 1, + refused: 1, + unchanged: 0, + skipped: 1, + failed: 1, + }) + expect(mocks.executeChild.mock.calls.map(([, opts]) => opts)).toEqual([ + { + workflowId: 'id/a/one', + args: [ + { owner: 'a', repo: 'one', syncNames: ['issues', 'forks'], direction: 'connectors' }, + ], + parentClosePolicy: 'ABANDON', + }, + { + workflowId: 'id/b/two', + args: [{ owner: 'b', repo: 'two', syncNames: ['issues'], direction: 'connectors' }], + parentClosePolicy: 'ABANDON', + }, + { + workflowId: 'id/c/three', + args: [ + { owner: 'c', repo: 'three', syncNames: ['pull-requests'], direction: 'connectors' }, + ], + parentClosePolicy: 'ABANDON', + }, + ]) + expect(mocks.executeChild.mock.calls.every(([wf]) => wf === 'switchSyncPipeline')).toBe(true) + }) +}) diff --git a/services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.ts b/services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.ts new file mode 100644 index 0000000000..9004157463 --- /dev/null +++ b/services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.ts @@ -0,0 +1,67 @@ +import { ParentClosePolicy, executeChild, log, proxyActivities } from '@temporalio/workflow' + +import type * as activities from '../activities/switchSyncPipelineActivities' +import type { ISweepRepo, ISwitchSyncResult } from '../syncPipelineSwitch' +import { switchSyncPipeline, switchSyncPipelineWorkflowId } from './switchSyncPipeline' + +const ALREADY_STARTED_ERROR = 'WorkflowExecutionAlreadyStartedError' + +const activity = proxyActivities({ + startToCloseTimeout: '2 minutes', + retry: { + maximumAttempts: 3, + backoffCoefficient: 2, + }, +}) + +export interface ISwitchSweepSummary { + repos: number + switched: number + refused: number + unchanged: number + skipped: number + failed: number +} + +export async function switchSyncPipelineSweep(): Promise { + const repos = await activity.listSwitchSweepRepos() + const summary: ISwitchSweepSummary = { + repos: repos.length, + switched: 0, + refused: 0, + unchanged: 0, + skipped: 0, + failed: 0, + } + + for (const repo of repos) { + const results = await switchRepo(repo) + if (results === 'skipped' || results === 'failed') { + summary[results] += 1 + continue + } + for (const result of results) { + summary[result.outcome] += 1 + } + } + + log.info('sync pipeline sweep complete', { ...summary }) + return summary +} + +async function switchRepo(repo: ISweepRepo): Promise { + try { + return await executeChild(switchSyncPipeline, { + workflowId: switchSyncPipelineWorkflowId(repo.owner, repo.repo), + args: [{ ...repo, direction: 'connectors' }], + parentClosePolicy: ParentClosePolicy.ABANDON, + }) + } catch (err) { + if (err instanceof Error && err.name === ALREADY_STARTED_ERROR) { + log.warn('sweep skipped a repo with a switch already running', { ...repo }) + return 'skipped' + } + log.error('sweep child switch failed', { ...repo, err }) + return 'failed' + } +} diff --git a/services/libs/data-access-layer/src/connectors/shadowRecords.ts b/services/libs/data-access-layer/src/connectors/shadowRecords.ts index 58705617f1..0ffdb7c235 100644 --- a/services/libs/data-access-layer/src/connectors/shadowRecords.ts +++ b/services/libs/data-access-layer/src/connectors/shadowRecords.ts @@ -92,47 +92,3 @@ export async function pruneMatchingShadowRecords( }, ) } - -export async function listShadowRecordsBatch( - qx: QueryExecutor, - unitId: string, - limit: number, -): Promise { - return qx.select( - `SELECT type, "sourceId", "occurredAt", data, "segmentId", "integrationId" - FROM integration.sync_shadow_records - WHERE "unitId" = $(unitId) - ORDER BY "occurredAt", type, "sourceId" - LIMIT $(limit)`, - { unitId, limit }, - ) -} - -export async function deleteShadowRecords( - qx: QueryExecutor, - unitId: string, - records: Pick[], -): Promise { - if (records.length === 0) { - return 0 - } - - return qx.result( - `DELETE FROM integration.sync_shadow_records sr - USING unnest( - $(types)::text[], $(sourceIds)::text[], $(segmentIds)::uuid[], $(integrationIds)::uuid[] - ) AS k(type, "sourceId", "segmentId", "integrationId") - WHERE sr."unitId" = $(unitId) - AND sr.type = k.type - AND sr."sourceId" = k."sourceId" - AND sr."segmentId" = k."segmentId" - AND sr."integrationId" = k."integrationId"`, - { - unitId, - types: records.map((r) => r.type), - sourceIds: records.map((r) => r.sourceId), - segmentIds: records.map((r) => r.segmentId), - integrationIds: records.map((r) => r.integrationId), - }, - ) -} diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts index 59ca8c42b8..79912c4a1f 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -4,12 +4,14 @@ import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common' import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' +import { recordShadowRecords } from './shadowRecords' import { claimDueUnits, listShadowDiffUnits, reassertUnitClaim, renewRunningUnitLease, reviveDeadLetteredSyncUnits, + takeOverSyncUnit, upsertSyncUnits, } from './syncUnits' @@ -412,3 +414,71 @@ describe('reassertUnitClaim', () => { expect((await readClaimState(qx, id)).lockedAgoSeconds).toBeNull() }) }) + +describe('takeOverSyncUnit', () => { + test('flips the unit to emit, rewinds a since walk, keeps a cursor walk, and drops every shadow row', async ({ + qx, + }) => { + const integrationId = await createIntegration(qx) + const sinceWalk = await createSyncUnit(qx, integrationId, { + channelName: 'repo', + syncName: 'issues', + watermark: { + phase: 'incremental', + since: '2026-10-19T09:00:00Z', + cursor: 'c1', + coveredWindows: [{ confirmedThrough: 'a', coveredUntil: 'b' }], + }, + }) + const cursorWalk = await createSyncUnit(qx, integrationId, { + channelName: 'repo', + syncName: 'forks', + watermark: { phase: 'incremental', since: '2026-10-19T09:00:00Z', cursor: 'c2' }, + }) + await recordShadowRecords(qx, sinceWalk, [ + { + type: 'issues-opened', + sourceId: 's1', + occurredAt: '2026-10-19T09:30:00Z', + data: {}, + segmentId: generateUUIDv1(), + integrationId, + }, + ]) + + expect(await takeOverSyncUnit(qx, sinceWalk, '2026-10-19T08:00:00.000Z')).toBe(1) + expect(await takeOverSyncUnit(qx, cursorWalk, null)).toBe(1) + expect(await takeOverSyncUnit(qx, sinceWalk, '2026-10-19T07:00:00.000Z')).toBe(0) + + const units = await qx.select( + `SELECT id, "emitEnabled", watermark + FROM integration.sync_units + WHERE id IN ($(sinceWalk), $(cursorWalk)) + ORDER BY "syncName"`, + { sinceWalk, cursorWalk }, + ) + expect(units).toEqual([ + { + id: cursorWalk, + emitEnabled: true, + watermark: { phase: 'incremental', since: '2026-10-19T09:00:00Z', cursor: 'c2' }, + }, + { + id: sinceWalk, + emitEnabled: true, + watermark: { + phase: 'incremental', + since: '2026-10-19T08:00:00.000Z', + cursor: null, + coveredWindows: [], + }, + }, + ]) + expect( + await qx.selectOne( + `SELECT count(*)::int AS n FROM integration.sync_shadow_records WHERE "unitId" = $(id)`, + { id: sinceWalk }, + ), + ).toEqual({ n: 0 }) + }) +}) diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index a729599187..cfb393251c 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -303,6 +303,30 @@ export async function setSyncUnitEmitEnabled( ) } +export async function takeOverSyncUnit( + qx: QueryExecutor, + id: string, + since: string | null, +): Promise { + return qx.tx(async (tx) => { + const flipped = await tx.result( + `UPDATE integration.sync_units + SET "emitEnabled" = true, + watermark = CASE + WHEN $(since)::text IS NULL THEN watermark + ELSE jsonb_build_object( + 'phase', 'incremental', 'since', $(since)::text, 'cursor', null, 'coveredWindows', '[]'::jsonb + ) + END, + "updatedAt" = now() + WHERE id = $(id) AND "emitEnabled" = false`, + { id, since }, + ) + await tx.result(`DELETE FROM integration.sync_shadow_records WHERE "unitId" = $(id)`, { id }) + return flipped + }) +} + export async function pauseSyncUnit(qx: QueryExecutor, id: string): Promise { return qx.result( `UPDATE integration.sync_units