From 86c46aa8ec02640c5906d457c00ba88919eb776e Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 20 Oct 2026 10:27:00 +0000 Subject: [PATCH 1/2] feat: nightly connector switch sweep with watermark-rewind handover Refs linuxfoundation/lfx-cdp-insights-ops#833 Signed-off-by: Mouad BANI --- .../apps/connectors_worker/src/activities.ts | 8 +- .../switchSyncPipelineActivities.test.ts | 84 ++++++++----------- .../switchSyncPipelineActivities.ts | 83 ++++++++++-------- .../src/bin/switch-sync-pipeline.ts | 3 +- services/apps/connectors_worker/src/main.ts | 2 + .../src/schedules/switchSyncPipelineSweep.ts | 31 +++++++ .../src/syncPipelineSwitch.test.ts | 32 +++++++ .../src/syncPipelineSwitch.ts | 37 +++++++- .../apps/connectors_worker/src/workflows.ts | 10 ++- .../src/workflows/switchSyncPipeline.test.ts | 49 ++++++++--- .../src/workflows/switchSyncPipeline.ts | 65 ++++++++------ .../workflows/switchSyncPipelineSweep.test.ts | 74 ++++++++++++++++ .../src/workflows/switchSyncPipelineSweep.ts | 67 +++++++++++++++ .../src/connectors/shadowRecords.ts | 44 ---------- .../src/connectors/syncUnits.test.ts | 70 ++++++++++++++++ .../src/connectors/syncUnits.ts | 24 ++++++ 16 files changed, 515 insertions(+), 168 deletions(-) create mode 100644 services/apps/connectors_worker/src/schedules/switchSyncPipelineSweep.ts create mode 100644 services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.test.ts create mode 100644 services/apps/connectors_worker/src/workflows/switchSyncPipelineSweep.ts diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts index de078c9e5e..06ed34583c 100644 --- a/services/apps/connectors_worker/src/activities.ts +++ b/services/apps/connectors_worker/src/activities.ts @@ -12,13 +12,15 @@ import { activateUnit, evaluateBackGate, evaluateForwardGate, + getNangoSyncLastRun, isUnitLocked, + listSwitchSweepRepos, pauseNangoRepoSync, quiesceUnit, - replayShadowRecords, resolveSwitchTarget, resumeNangoRepoSync, setUnitEmitEnabled, + takeOverUnit, } from './activities/switchSyncPipelineActivities' import { executeSync } from './activities/syncRunActivities' @@ -30,17 +32,19 @@ export { evaluateBackGate, evaluateForwardGate, executeSync, + getNangoSyncLastRun, isUnitLocked, listShadowDiffChannels, + listSwitchSweepRepos, logDispatchSummary, pauseNangoRepoSync, quiesceUnit, - replayShadowRecords, 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..40514043a8 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, + getNangoSyncLastRun, pauseNangoRepoSync, - replayShadowRecords, 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}`, - ) - - await expect(replayShadowRecords({ unitId: 'u-1' })).resolves.toBe(3) + it('reads the last nango run of the one sync, then takes a unit over an hour before it, or keeps a cursor walk', async () => { + mocks.getNangoConnectionStatus.mockResolvedValue([ + { name: 'issues', status: 'PAUSED', finishedAt: '2026-10-19T09:00:00.000Z' }, + { name: 'forks', status: 'SUCCESS' }, + ]) + await expect( + getNangoSyncLastRun({ connectionId: 'conn-1', nangoSync: 'issues' }), + ).resolves.toEqual({ finishedAt: '2026-10-19T09:00:00.000Z' }) + await expect( + getNangoSyncLastRun({ connectionId: 'conn-1', nangoSync: 'forks' }), + ).resolves.toEqual({ finishedAt: null }) + expect(mocks.getNangoConnectionStatus).toHaveBeenCalledWith('github', 'conn-1') - expect(mocks.publishExternalResult).toHaveBeenCalledTimes(3) - expect(mocks.publishExternalResult).toHaveBeenCalledWith('int-1', { - type: IntegrationResultType.ACTIVITY, - segmentId: 'seg-1', - data: { sourceId: 'c' }, - }) - 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) + mocks.takeOverSyncUnit.mockResolvedValue(1) + await expect( + takeOverUnit({ + unitId: 'u-1', + syncName: 'issues', + lastRunFinishedAt: '2026-10-19T09:00:00.000Z', + }), + ).resolves.toEqual({ since: '2026-10-19T08:00:00.000Z' }) + expect(mocks.takeOverSyncUnit).toHaveBeenCalledWith('qx', 'u-1', '2026-10-19T08:00:00.000Z') + await expect( + takeOverUnit({ + unitId: 'u-2', + syncName: 'forks', + lastRunFinishedAt: '2026-10-19T09:00:00.000Z', + }), + ).resolves.toEqual({ since: null }) + 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..d75cf29257 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,15 @@ import { evaluateDiffCleanliness, evaluateForwardUnitGate, evaluateNangoConsumption, + groupUnitsForSweep, + 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 +79,20 @@ export interface IUnitInput { unitId: string } +export interface INangoSyncLastRun { + finishedAt: string | null +} + +export interface ITakeOverUnitInput { + unitId: string + syncName: string + lastRunFinishedAt: string | null +} + +export interface ITakeOverUnitResult { + since: string | null +} + export async function resolveSwitchTarget( input: IResolveSwitchTargetInput, ): Promise { @@ -191,31 +203,34 @@ 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 getNangoSyncLastRun(input: INangoSyncInput): Promise { + await initNangoCloudClient() + const syncs = await getNangoConnectionStatus(NangoIntegration.GITHUB, input.connectionId) + const sync = syncs?.find((s) => s.name === input.nangoSync) + const finishedAt = sync?.finishedAt ?? null + svc.log.info( + { event: SWITCH_EVENT, ...input, finishedAt, status: sync?.status ?? null }, + 'nango sync last run read', + ) + return { finishedAt } +} + +export async function takeOverUnit(input: ITakeOverUnitInput): Promise { + const since = handoverSince(input.syncName, input.lastRunFinishedAt) + 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 } +} - svc.log.info({ event: SWITCH_EVENT, ...input, replayed }, 'shadow records replayed') - return replayed +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..254bf7a5fc 100644 --- a/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts +++ b/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts @@ -8,6 +8,8 @@ import { evaluateDiffCleanliness, evaluateForwardUnitGate, evaluateNangoConsumption, + groupUnitsForSweep, + handoverSince, recentDiffedDays, resolveSwitchTargets, } from './syncPipelineSwitch' @@ -109,3 +111,33 @@ 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('handoverSince', () => { + it('rewinds an hour before the last nango run, except for cursor-walked forks', () => { + expect(handoverSince('issues', '2026-10-19T09:00:00Z')).toBe('2026-10-19T08:00:00.000Z') + expect(handoverSince('forks', '2026-10-19T09:00:00Z')).toBeNull() + expect(handoverSince('issues', null)).toBeNull() + }) +}) diff --git a/services/apps/connectors_worker/src/syncPipelineSwitch.ts b/services/apps/connectors_worker/src/syncPipelineSwitch.ts index b3ce643339..3f6a41ea25 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,19 @@ 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 + +const CURSOR_WALKED_SYNCS = new Set(['forks']) export function resolveSwitchTargets(requested?: string[]): ISwitchSyncTarget[] { const syncNames = requested?.length ? requested : GITHUB_SYNC_MAPPINGS.map((m) => m.connectorSync) @@ -124,3 +138,24 @@ 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 handoverSince(syncName: string, lastRunFinishedAt: string | null): string | null { + if (lastRunFinishedAt === null || CURSOR_WALKED_SYNCS.has(syncName)) { + return null + } + return new Date(new Date(lastRunFinishedAt).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..758c9724fc 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(), + getNangoSyncLastRun: vi.fn(), + takeOverUnit: vi.fn(), })) vi.mock('@temporalio/workflow', () => ({ @@ -51,10 +52,13 @@ describe('switchSyncPipeline', () => { activities.pauseNangoRepoSync.mockResolvedValue(undefined) activities.resumeNangoRepoSync.mockResolvedValue(undefined) activities.setUnitEmitEnabled.mockResolvedValue(undefined) - activities.replayShadowRecords.mockResolvedValue(3) + activities.getNangoSyncLastRun.mockResolvedValue({ finishedAt: '2026-10-19T09:00:00Z' }) + activities.takeOverUnit.mockImplementation(async ({ syncName }) => ({ + since: syncName === 'forks' ? null : '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 last nango run, 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 +70,31 @@ 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.getNangoSyncLastRun, 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.takeOverUnit).toHaveBeenCalledWith({ unitId: 'u-issues', - emitEnabled: true, + syncName: 'issues', + lastRunFinishedAt: '2026-10-19T09:00:00Z', }) expect(activities.activateUnit.mock.calls).toEqual([ [{ unitId: 'u-issues' }], @@ -92,7 +102,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, and an emitting unit that gets healed', async () => { activities.resolveSwitchTarget.mockResolvedValue({ target, syncs: [ @@ -100,9 +110,13 @@ describe('switchSyncPipeline', () => { { syncName: 'stars', nangoSync: null }, sync('forks'), sync('pull-requests'), + sync('issue-comments'), sync('discussions'), ], }) + activities.getNangoSyncLastRun.mockImplementation(async ({ nangoSync }) => ({ + finishedAt: nangoSync === 'issue-comments' ? null : '2026-10-19T09:00:00Z', + })) const seen: Record = {} activities.evaluateForwardGate.mockImplementation(async ({ syncName }) => { seen[syncName] = (seen[syncName] ?? 0) + 1 @@ -131,10 +145,15 @@ 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: 'discussions', outcome: 'unchanged', - reason: 'already emitting; 3 shadow records replayed', + reason: 'already emitting; shadow records cleared', }, ]) expect(activities.pauseNangoRepoSync).toHaveBeenCalledTimes(1) @@ -143,10 +162,16 @@ describe('switchSyncPipeline', () => { nangoSync: 'discussions', }) expect(activities.setUnitEmitEnabled).not.toHaveBeenCalled() - expect(activities.replayShadowRecords).toHaveBeenCalledWith({ unitId: 'u-discussions' }) + expect(activities.takeOverUnit).toHaveBeenCalledTimes(1) + expect(activities.takeOverUnit).toHaveBeenCalledWith({ + unitId: 'u-discussions', + syncName: 'discussions', + lastRunFinishedAt: null, + }) expect(activities.activateUnit.mock.calls).toEqual([ [{ unitId: 'u-forks' }], [{ unitId: 'u-pull-requests' }], + [{ unitId: 'u-issue-comments' }], [{ 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..eb22659438 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 replayed: number try { - replayed = await replayActivity.replayShadowRecords({ unitId }) + await takeOverActivity.takeOverUnit({ + unitId, + syncName: sync.syncName, + lastRunFinishedAt: null, + }) } catch (err) { - return refused( - sync, - `emitting but shadow replay failed: ${describeFailure(err)}; rerun to finish`, - ) + return refused(sync, `emitting but shadow cleanup failed: ${describeFailure(err)}; rerun`) } await activity.activateUnit({ unitId }) - - const detail = `${replayed} shadow records replayed` - return { syncName: sync.syncName, outcome, reason: reason ? `${reason}; ${detail}` : detail } + 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 From 304f85f45a704c2df7d71c2039c93f35ce522731 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 20 Oct 2026 10:38:04 +0000 Subject: [PATCH 2/2] fix: hand over from the consumed nango boundary and roll back a failed takeover Refs linuxfoundation/lfx-cdp-insights-ops#833 Signed-off-by: Mouad BANI --- .../apps/connectors_worker/src/activities.ts | 4 +- .../switchSyncPipelineActivities.test.ts | 42 ++++++------- .../switchSyncPipelineActivities.ts | 41 +++++++++--- .../src/syncPipelineSwitch.test.ts | 17 +++-- .../src/syncPipelineSwitch.ts | 15 +++-- .../src/workflows/switchSyncPipeline.test.ts | 63 +++++++++++++------ .../src/workflows/switchSyncPipeline.ts | 51 ++++++++++----- 7 files changed, 153 insertions(+), 80 deletions(-) diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts index 06ed34583c..fad4b0612b 100644 --- a/services/apps/connectors_worker/src/activities.ts +++ b/services/apps/connectors_worker/src/activities.ts @@ -12,11 +12,11 @@ import { activateUnit, evaluateBackGate, evaluateForwardGate, - getNangoSyncLastRun, isUnitLocked, listSwitchSweepRepos, pauseNangoRepoSync, quiesceUnit, + readNangoHandoverPoint, resolveSwitchTarget, resumeNangoRepoSync, setUnitEmitEnabled, @@ -32,13 +32,13 @@ export { evaluateBackGate, evaluateForwardGate, executeSync, - getNangoSyncLastRun, isUnitLocked, listShadowDiffChannels, listSwitchSweepRepos, logDispatchSummary, pauseNangoRepoSync, quiesceUnit, + readNangoHandoverPoint, resolveSwitchTarget, resumeNangoRepoSync, runOctolensDiscovery, diff --git a/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts b/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts index 40514043a8..c404bb7ffa 100644 --- a/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.test.ts @@ -2,8 +2,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { evaluateForwardGate, - getNangoSyncLastRun, pauseNangoRepoSync, + readNangoHandoverPoint, resolveSwitchTarget, setUnitEmitEnabled, takeOverUnit, @@ -147,35 +147,35 @@ describe('switch activities', () => { ) }) - it('reads the last nango run of the one sync, then takes a unit over an hour before it, or keeps a cursor walk', async () => { + 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( - getNangoSyncLastRun({ connectionId: 'conn-1', nangoSync: 'issues' }), - ).resolves.toEqual({ finishedAt: '2026-10-19T09:00:00.000Z' }) - await expect( - getNangoSyncLastRun({ connectionId: 'conn-1', nangoSync: 'forks' }), - ).resolves.toEqual({ finishedAt: null }) + 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 }) mocks.takeOverSyncUnit.mockResolvedValue(1) await expect( - takeOverUnit({ - unitId: 'u-1', - syncName: 'issues', - lastRunFinishedAt: '2026-10-19T09:00:00.000Z', - }), - ).resolves.toEqual({ since: '2026-10-19T08:00:00.000Z' }) - expect(mocks.takeOverSyncUnit).toHaveBeenCalledWith('qx', 'u-1', '2026-10-19T08:00:00.000Z') - await expect( - takeOverUnit({ - unitId: 'u-2', - syncName: 'forks', - lastRunFinishedAt: '2026-10-19T09:00:00.000Z', - }), - ).resolves.toEqual({ since: null }) + 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.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 d75cf29257..3181074654 100644 --- a/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.ts +++ b/services/apps/connectors_worker/src/activities/switchSyncPipelineActivities.ts @@ -30,6 +30,7 @@ import { evaluateForwardUnitGate, evaluateNangoConsumption, groupUnitsForSweep, + handoverPoint, handoverSince, recentDiffedDays, resolveSwitchTargets, @@ -79,14 +80,20 @@ export interface IUnitInput { unitId: string } -export interface INangoSyncLastRun { - finishedAt: string | null +export interface IHandoverPointInput { + integrationId: string + connectionId: string + syncName: string + nangoSync: string +} + +export interface IHandoverPoint { + handoverAt: string | null } export interface ITakeOverUnitInput { unitId: string - syncName: string - lastRunFinishedAt: string | null + handoverAt: string | null } export interface ITakeOverUnitResult { @@ -203,20 +210,34 @@ export async function activateUnit(input: IUnitInput): Promise { svc.log.info({ event: SWITCH_EVENT, ...input, activated: activated === 1 }, 'sync unit activated') } -export async function getNangoSyncLastRun(input: INangoSyncInput): Promise { +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 finishedAt = sync?.finishedAt ?? null + 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, finishedAt, status: sync?.status ?? null }, - 'nango sync last run read', + { + event: SWITCH_EVENT, + ...input, + nangoStatus: sync?.status ?? null, + nangoRunFinishedAt, + consumedAt, + handoverAt, + }, + 'nango handover point read', ) - return { finishedAt } + return { handoverAt } } export async function takeOverUnit(input: ITakeOverUnitInput): Promise { - const since = handoverSince(input.syncName, input.lastRunFinishedAt) + 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 }, diff --git a/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts b/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts index 254bf7a5fc..21d53e66a8 100644 --- a/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts +++ b/services/apps/connectors_worker/src/syncPipelineSwitch.test.ts @@ -9,6 +9,7 @@ import { evaluateForwardUnitGate, evaluateNangoConsumption, groupUnitsForSweep, + handoverPoint, handoverSince, recentDiffedDays, resolveSwitchTargets, @@ -134,10 +135,16 @@ describe('groupUnitsForSweep', () => { }) }) -describe('handoverSince', () => { - it('rewinds an hour before the last nango run, except for cursor-walked forks', () => { - expect(handoverSince('issues', '2026-10-19T09:00:00Z')).toBe('2026-10-19T08:00:00.000Z') - expect(handoverSince('forks', '2026-10-19T09:00:00Z')).toBeNull() - expect(handoverSince('issues', null)).toBeNull() +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 3f6a41ea25..65b8aa44f1 100644 --- a/services/apps/connectors_worker/src/syncPipelineSwitch.ts +++ b/services/apps/connectors_worker/src/syncPipelineSwitch.ts @@ -41,8 +41,6 @@ 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 -const CURSOR_WALKED_SYNCS = new Set(['forks']) - export function resolveSwitchTargets(requested?: string[]): ISwitchSyncTarget[] { const syncNames = requested?.length ? requested : GITHUB_SYNC_MAPPINGS.map((m) => m.connectorSync) return syncNames.map((syncName) => ({ @@ -153,9 +151,16 @@ export function groupUnitsForSweep(units: IShadowDiffUnit[]): ISweepRepo[] { return [...byRepo.values()] } -export function handoverSince(syncName: string, lastRunFinishedAt: string | null): string | null { - if (lastRunFinishedAt === null || CURSOR_WALKED_SYNCS.has(syncName)) { +export function handoverPoint( + nangoRunFinishedAt: string | null, + consumedAt: string | null, +): string | null { + if (!nangoRunFinishedAt || !consumedAt) { return null } - return new Date(new Date(lastRunFinishedAt).getTime() - NANGO_HANDOVER_OVERLAP_MS).toISOString() + 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/switchSyncPipeline.test.ts b/services/apps/connectors_worker/src/workflows/switchSyncPipeline.test.ts index 758c9724fc..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,7 @@ const activities = vi.hoisted(() => ({ pauseNangoRepoSync: vi.fn(), resumeNangoRepoSync: vi.fn(), setUnitEmitEnabled: vi.fn(), - getNangoSyncLastRun: vi.fn(), + readNangoHandoverPoint: vi.fn(), takeOverUnit: vi.fn(), })) @@ -52,13 +52,11 @@ describe('switchSyncPipeline', () => { activities.pauseNangoRepoSync.mockResolvedValue(undefined) activities.resumeNangoRepoSync.mockResolvedValue(undefined) activities.setUnitEmitEnabled.mockResolvedValue(undefined) - activities.getNangoSyncLastRun.mockResolvedValue({ finishedAt: '2026-10-19T09:00:00Z' }) - activities.takeOverUnit.mockImplementation(async ({ syncName }) => ({ - since: syncName === 'forks' ? null : '2026-10-19T08:00:00.000Z', - })) + 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, reads the last nango run, pauses nango, takes the unit over, 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 @@ -80,7 +78,7 @@ describe('switchSyncPipeline', () => { const order = [ activities.quiesceUnit, activities.isUnitLocked, - activities.getNangoSyncLastRun, + activities.readNangoHandoverPoint, activities.pauseNangoRepoSync, activities.takeOverUnit, activities.activateUnit, @@ -91,10 +89,14 @@ describe('switchSyncPipeline', () => { expect(activities.isUnitLocked).toHaveBeenCalledTimes(3) 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', - syncName: 'issues', - lastRunFinishedAt: '2026-10-19T09:00:00Z', + handoverAt: '2026-10-19T09:00:00Z', }) expect(activities.activateUnit.mock.calls).toEqual([ [{ unitId: 'u-issues' }], @@ -102,7 +104,7 @@ describe('switchSyncPipeline', () => { ]) }) - it('forward: refusals, a stuck lock, a unit that changed while draining, a nango sync that never ran, 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: [ @@ -111,12 +113,19 @@ describe('switchSyncPipeline', () => { sync('forks'), sync('pull-requests'), sync('issue-comments'), + sync('pull-request-comments'), + sync('pull-request-commits'), sync('discussions'), ], }) - activities.getNangoSyncLastRun.mockImplementation(async ({ nangoSync }) => ({ - finishedAt: nangoSync === 'issue-comments' ? null : '2026-10-19T09:00:00Z', - })) + 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 @@ -150,28 +159,42 @@ describe('switchSyncPipeline', () => { 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; 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.takeOverUnit).toHaveBeenCalledTimes(1) - expect(activities.takeOverUnit).toHaveBeenCalledWith({ + expect(activities.takeOverUnit).toHaveBeenCalledTimes(2) + expect(activities.takeOverUnit).toHaveBeenLastCalledWith({ unitId: 'u-discussions', - syncName: 'discussions', - lastRunFinishedAt: null, + 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 eb22659438..615969b1b4 100644 --- a/services/apps/connectors_worker/src/workflows/switchSyncPipeline.ts +++ b/services/apps/connectors_worker/src/workflows/switchSyncPipeline.ts @@ -100,11 +100,14 @@ async function switchToConnectors( return refused(sync, `unit changed while draining: ${drainedGate.reason}`) } - const lastRun = await activity.getNangoSyncLastRun({ - connectionId: target.connectionId, - nangoSync: sync.nangoSync, - }) - if (!lastRun.finishedAt) { + let handover: activities.IHandoverPoint + try { + handover = await activity.readNangoHandoverPoint({ ...gateInput, nangoSync: sync.nangoSync }) + } catch (err) { + await activity.activateUnit({ unitId: gate.unitId }) + return refused(sync, `nango status read failed: ${describeFailure(err)}`) + } + if (!handover.handoverAt) { await activity.activateUnit({ unitId: gate.unitId }) return refused(sync, 'nango sync has no completed run to hand over from') } @@ -123,13 +126,14 @@ async function switchToConnectors( try { takeover = await takeOverActivity.takeOverUnit({ unitId: gate.unitId, - syncName: sync.syncName, - lastRunFinishedAt: lastRun.finishedAt, + handoverAt: handover.handoverAt, }) } catch (err) { - return refused( + return abortAfterNangoPause( + target, sync, - `nango paused but unit takeover failed: ${describeFailure(err)}; rerun to retry`, + gate.unitId, + `unit takeover failed: ${describeFailure(err)}`, ) } @@ -137,12 +141,29 @@ async function switchToConnectors( return { syncName: sync.syncName, outcome: 'switched', - reason: takeover.since - ? `since rewound to ${takeover.since}` - : 'cursor-walked sync, watermark kept', + reason: `since rewound to ${takeover.since}`, } } +async function abortAfterNangoPause( + target: ISwitchRepoTarget, + sync: ISwitchSyncTarget, + unitId: string, + reason: string, +): Promise { + let nangoState = 'nango resumed' + try { + await activity.resumeNangoRepoSync({ + connectionId: target.connectionId, + nangoSync: sync.nangoSync, + }) + } catch (err) { + nangoState = `nango still paused, resume failed: ${describeFailure(err)}` + } + await activity.activateUnit({ unitId }) + return refused(sync, `${reason}; ${nangoState}; unit back in shadow`) +} + async function healEmittingUnit( target: ISwitchRepoTarget, sync: ISwitchSyncTarget, @@ -159,11 +180,7 @@ async function healEmittingUnit( } try { - await takeOverActivity.takeOverUnit({ - unitId, - syncName: sync.syncName, - lastRunFinishedAt: null, - }) + await takeOverActivity.takeOverUnit({ unitId, handoverAt: null }) } catch (err) { return refused(sync, `emitting but shadow cleanup failed: ${describeFailure(err)}; rerun`) }