From 6149759a6d8c419c65ed9f6b2cfff0b7fff6d025 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:27:11 +0200 Subject: [PATCH 01/13] refactor: move the sync run workflow id into a module scripts can import MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/activities/dispatcherActivities.ts | 16 +--------------- .../src/syncRunWorkflowId.ts | 19 +++++++++++++++++++ 2 files changed, 20 insertions(+), 15 deletions(-) create mode 100644 services/apps/connectors_worker/src/syncRunWorkflowId.ts diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts index 2a563e2dd0..6052e1fae5 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -7,13 +7,13 @@ import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal import { svc } from '../main' import { runningProbeRunAt, shortDeferRunAt } from '../scheduling' +import { syncRunWorkflowId } from '../syncRunWorkflowId' import type { IAdmissionResult, IDispatchCounts, StartRunResult } from '../types' const TASK_QUEUE = 'connectors' const HEARTBEAT_TTL_SECONDS = 300 const DEFAULT_RUN_ESTIMATE = 50 const BUDGET_PROBE_CONCURRENCY = 10 -const CHANNEL_LABEL_MAX_LENGTH = 80 export async function claimDue(limit: number): Promise { return claimDueUnits(dbStoreQx(svc.postgres.writer), limit) @@ -44,20 +44,6 @@ export async function deferUnit(unitId: string): Promise { await rescheduleUnit(dbStoreQx(svc.postgres.writer), unitId, shortDeferRunAt()) } -function channelLabel(channelName: string): string { - const label = channelName - .replace(/^[a-z][a-z0-9+.-]*:\/\/[^/]+\//i, '') - .replace(/^\/+|\/+$/g, '') - .replace(/\s+/g, '-') - .slice(0, CHANNEL_LABEL_MAX_LENGTH) - - return label || 'unknown-channel' -} - -export function syncRunWorkflowId(unit: IClaimedUnit): string { - return `sync-run/${unit.platform}/${unit.syncName}/${channelLabel(unit.channelName)}/${unit.id}` -} - export async function startRun(unit: IClaimedUnit): Promise { try { await svc.temporal.workflow.start('syncRun', { diff --git a/services/apps/connectors_worker/src/syncRunWorkflowId.ts b/services/apps/connectors_worker/src/syncRunWorkflowId.ts new file mode 100644 index 0000000000..471587ecd8 --- /dev/null +++ b/services/apps/connectors_worker/src/syncRunWorkflowId.ts @@ -0,0 +1,19 @@ +import type { IClaimedUnit } from '@crowd/data-access-layer/src/connectors' + +const CHANNEL_LABEL_MAX_LENGTH = 80 + +function channelLabel(channelName: string): string { + const label = channelName + .replace(/^[a-z][a-z0-9+.-]*:\/\/[^/]+\//i, '') + .replace(/^\/+|\/+$/g, '') + .replace(/\s+/g, '-') + .slice(0, CHANNEL_LABEL_MAX_LENGTH) + + return label || 'unknown-channel' +} + +export function syncRunWorkflowId( + unit: Pick, +): string { + return `sync-run/${unit.platform}/${unit.syncName}/${channelLabel(unit.channelName)}/${unit.id}` +} From 14f855e1d16071ff431c19fcc72f44f7df0602a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:27:14 +0200 Subject: [PATCH 02/13] feat: add guarded octolens mention poll watermark writes to the dal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../data-access-layer/src/octolens/index.ts | 141 +++++++++++++ .../src/octolens/mentionPollWatermark.test.ts | 194 ++++++++++++++++++ 2 files changed, 335 insertions(+) create mode 100644 services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts index 496d1a9284..1ed15f9b2f 100644 --- a/services/libs/data-access-layer/src/octolens/index.ts +++ b/services/libs/data-access-layer/src/octolens/index.ts @@ -273,3 +273,144 @@ export async function reconcileOctolensKeywordMappings( return { deactivated, created, restored, softDeleted } }) } + +const MENTION_POLL_SYNC_NAME = 'mentionPoll' + +export interface IOctolensMentionPollUnit { + id: string + integrationId: string + channelName: string + status: string + lockedAt: Date | null + watermark: Record | null +} + +export type OctolensMentionPollWriteOutcome = 'applied' | 'locked' | 'inactive' | 'missing' + +export interface IOctolensMentionPollWriteResult { + outcome: OctolensMentionPollWriteOutcome + watermark: Record | null +} + +export async function listOctolensAnchorIntegrationIds(qx: QueryExecutor): Promise { + const rows: { id: string }[] = await qx.select( + ` + SELECT i.id + FROM public.integrations i + WHERE i.platform = $(platform) + AND i."segmentId" IS NULL + AND i."deletedAt" IS NULL + ORDER BY i.id + `, + { platform: PlatformType.OCTOLENS }, + ) + return rows.map((r) => r.id) +} + +export async function listOctolensMentionPollUnits( + qx: QueryExecutor, +): Promise { + return qx.select( + ` + SELECT su.id, su."integrationId", su."channelName", su.status, su."lockedAt", su.watermark + FROM integration.sync_units su + JOIN public.integrations i ON i.id = su."integrationId" + WHERE su.platform = $(platform) + AND su."syncName" = $(syncName) + AND i.platform = $(platform) + AND i."segmentId" IS NULL + AND i."deletedAt" IS NULL + ORDER BY su.id + `, + { platform: PlatformType.OCTOLENS, syncName: MENTION_POLL_SYNC_NAME }, + ) +} + +const STORED_WATERMARK = `(CASE WHEN jsonb_typeof(su.watermark) = 'object' THEN su.watermark ELSE '{}'::jsonb END)` + +const STORED_BACKFILL = `(CASE WHEN jsonb_typeof(su.watermark->'backfill') = 'array' THEN su.watermark->'backfill' ELSE '[]'::jsonb END)` + +// A run keeps the watermark in memory and overwrites it when it ends, so this only writes while +// the unit is unclaimed; callers must separately rule out a run still in flight with no claim. +async function updateIdleMentionPollWatermark( + qx: QueryExecutor, + unitId: string, + watermarkSql: string, + params: Record, +): Promise { + const filter = { unitId, platform: PlatformType.OCTOLENS, syncName: MENTION_POLL_SYNC_NAME } + const updated: { watermark: Record }[] = await qx.select( + ` + UPDATE integration.sync_units su + SET watermark = ${watermarkSql}, "updatedAt" = now() + WHERE su.id = $(unitId) + AND su.platform = $(platform) + AND su."syncName" = $(syncName) + AND su.status = 'active' + AND su."lockedAt" IS NULL + RETURNING su.watermark + `, + { ...params, ...filter }, + ) + if (updated.length > 0) { + return { outcome: 'applied', watermark: updated[0].watermark } + } + + const unit: { status: string; watermark: Record | null } | null = + await qx.selectOneOrNone( + ` + SELECT su.status, su.watermark + FROM integration.sync_units su + WHERE su.id = $(unitId) AND su.platform = $(platform) AND su."syncName" = $(syncName) + `, + filter, + ) + if (!unit) { + return { outcome: 'missing', watermark: null } + } + return { outcome: unit.status === 'active' ? 'locked' : 'inactive', watermark: unit.watermark } +} + +export async function queueOctolensKeywordBackfills( + qx: QueryExecutor, + unitId: string, + keywordIds: number[], +): Promise { + return updateIdleMentionPollWatermark( + qx, + unitId, + `jsonb_set( + ${STORED_WATERMARK}, + '{backfill}', + ${STORED_BACKFILL} || COALESCE( + ( + SELECT jsonb_agg(jsonb_build_object('keywordId', k.id, 'cursor', NULL) ORDER BY k.ord) + FROM unnest($(keywordIds)::int[]) WITH ORDINALITY AS k(id, ord) + WHERE NOT EXISTS ( + SELECT 1 FROM jsonb_array_elements(${STORED_BACKFILL}) e + WHERE e->'keywordId' = to_jsonb(k.id) + ) + ), + '[]'::jsonb + ) + )`, + { keywordIds: [...new Set(keywordIds)] }, + ) +} + +export async function rewindOctolensMentionPollStartDate( + qx: QueryExecutor, + unitId: string, + startDate: string, +): Promise { + return updateIdleMentionPollWatermark( + qx, + unitId, + `${STORED_WATERMARK} || jsonb_build_object( + 'startDate', $(startDate)::text, + 'cursor', NULL, + 'pendingMaxTimestamp', NULL + )`, + { startDate }, + ) +} diff --git a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts new file mode 100644 index 0000000000..827005b2b2 --- /dev/null +++ b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts @@ -0,0 +1,194 @@ +import { test as base, describe, expect } from 'vitest' + +import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common' +import type { QueryExecutor } from '@crowd/database' +import { withQx } from '@crowd/test-kit/db' + +import { + listOctolensAnchorIntegrationIds, + listOctolensMentionPollUnits, + queueOctolensKeywordBackfills, + rewindOctolensMentionPollStartDate, +} from './index' + +const test = withQx(base) + +async function createAnchorIntegration( + qx: QueryExecutor, + overrides: { deletedAt?: string } = {}, +): Promise { + const id = generateUUIDv1() + await qx.result( + ` + INSERT INTO public.integrations (id, platform, status, "tenantId", "createdAt", "updatedAt", "deletedAt") + VALUES ($(id), 'octolens', 'done', $(tenantId), NOW(), NOW(), $(deletedAt)::timestamptz) + `, + { id, tenantId: DEFAULT_TENANT_ID, deletedAt: overrides.deletedAt ?? null }, + ) + return id +} + +async function createMentionPollUnit( + qx: QueryExecutor, + integrationId: string, + overrides: { + status?: string + locked?: boolean + watermark?: Record | null + } = {}, +): Promise { + const id = generateUUIDv1() + await qx.result( + ` + INSERT INTO integration.sync_units + (id, "integrationId", platform, "channelId", "channelName", "syncName", status, "emitEnabled", watermark, "lockedAt") + VALUES + ($(id), $(integrationId), 'octolens', 'octolens-mentions', 'octolens-mentions', 'mentionPoll', + $(status), true, $(watermark)::jsonb, CASE WHEN $(locked) THEN now() ELSE NULL END) + `, + { + id, + integrationId, + status: overrides.status ?? 'active', + locked: overrides.locked ?? false, + watermark: overrides.watermark == null ? null : JSON.stringify(overrides.watermark), + }, + ) + return id +} + +async function readWatermark(qx: QueryExecutor, unitId: string): Promise { + const row: { watermark: unknown } = await qx.selectOne( + `SELECT watermark FROM integration.sync_units WHERE id = $(unitId)`, + { unitId }, + ) + return row.watermark +} + +const MID_CYCLE = { + startDate: '2026-10-16T10:00:00.000Z', + cursor: 'c-2', + pendingMaxTimestamp: '2026-10-16T10:04:00.000Z', + backfill: [{ keywordId: 7, cursor: 'k7-page3' }], +} + +describe('listOctolensMentionPollUnits', () => { + test('returns only mention-poll units of live segment-less octolens integrations', async ({ + qx, + }) => { + const anchorId = await createAnchorIntegration(qx) + const unitId = await createMentionPollUnit(qx, anchorId, { status: 'dead_letter' }) + const deletedAnchorId = await createAnchorIntegration(qx, { deletedAt: '2026-01-01' }) + await createMentionPollUnit(qx, deletedAnchorId) + + expect(await listOctolensAnchorIntegrationIds(qx)).toEqual([anchorId]) + const units = await listOctolensMentionPollUnits(qx) + expect(units.map((u) => [u.id, u.integrationId, u.status])).toEqual([ + [unitId, anchorId, 'dead_letter'], + ]) + }) +}) + +describe('queueOctolensKeywordBackfills', () => { + test('appends new keywords after the existing queue and skips already-queued ones', async ({ + qx, + }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + watermark: MID_CYCLE, + }) + + const result = await queueOctolensKeywordBackfills(qx, unitId, [9, 7, 8, 9]) + + expect(result.outcome).toBe('applied') + expect(await readWatermark(qx, unitId)).toEqual({ + ...MID_CYCLE, + backfill: [ + { keywordId: 7, cursor: 'k7-page3' }, + { keywordId: 9, cursor: null }, + { keywordId: 8, cursor: null }, + ], + }) + }) + + test('creates the queue on a unit that has never run', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx)) + + await queueOctolensKeywordBackfills(qx, unitId, [5]) + + expect(await readWatermark(qx, unitId)).toEqual({ backfill: [{ keywordId: 5, cursor: null }] }) + }) + + test('refuses and writes nothing while the unit is claimed', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + locked: true, + watermark: MID_CYCLE, + }) + + const result = await queueOctolensKeywordBackfills(qx, unitId, [9]) + + expect(result.outcome).toBe('locked') + expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE) + }) + + test('refuses a unit that is not active', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + status: 'dead_letter', + watermark: MID_CYCLE, + }) + + const result = await queueOctolensKeywordBackfills(qx, unitId, [9]) + + expect(result.outcome).toBe('inactive') + expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE) + }) + + test('reports a unit id that is not a mention-poll unit as missing', async ({ qx }) => { + const result = await queueOctolensKeywordBackfills(qx, generateUUIDv1(), [9]) + + expect(result.outcome).toBe('missing') + }) +}) + +describe('rewindOctolensMentionPollStartDate', () => { + test('sets the start date, clears the cycle progress and keeps the backfill queue', async ({ + qx, + }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + watermark: MID_CYCLE, + }) + + const result = await rewindOctolensMentionPollStartDate(qx, unitId, '2026-09-01T00:00:00.000Z') + + expect(result.outcome).toBe('applied') + expect(await readWatermark(qx, unitId)).toEqual({ + startDate: '2026-09-01T00:00:00.000Z', + cursor: null, + pendingMaxTimestamp: null, + backfill: MID_CYCLE.backfill, + }) + }) + + test('refuses and writes nothing while the unit is claimed', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + locked: true, + watermark: MID_CYCLE, + }) + + const result = await rewindOctolensMentionPollStartDate(qx, unitId, '2026-09-01T00:00:00.000Z') + + expect(result.outcome).toBe('locked') + expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE) + }) + + test('refuses a unit that is not active', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + status: 'paused', + watermark: MID_CYCLE, + }) + + const result = await rewindOctolensMentionPollStartDate(qx, unitId, '2026-09-01T00:00:00.000Z') + + expect(result.outcome).toBe('inactive') + expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE) + }) +}) From 1412695bfab96e1b6b2c24d6f823c7d318a195b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:27:16 +0200 Subject: [PATCH 03/13] feat: add a script that queues octolens keyword backfills or rewinds the poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- services/apps/connectors_worker/package.json | 3 +- .../src/bin/backfill-octolens-mentions.ts | 151 +++++++++++++++++ .../src/octolensBackfill.test.ts | 133 +++++++++++++++ .../connectors_worker/src/octolensBackfill.ts | 156 ++++++++++++++++++ 4 files changed, 442 insertions(+), 1 deletion(-) create mode 100644 services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts create mode 100644 services/apps/connectors_worker/src/octolensBackfill.test.ts create mode 100644 services/apps/connectors_worker/src/octolensBackfill.ts diff --git a/services/apps/connectors_worker/package.json b/services/apps/connectors_worker/package.json index 6fb4787107..905a3271a1 100644 --- a/services/apps/connectors_worker/package.json +++ b/services/apps/connectors_worker/package.json @@ -10,7 +10,8 @@ "script:seed-github-sync-units": "tsx src/bin/seed-github-sync-units.ts", "script:audit-github-app-budget": "tsx src/bin/audit-github-app-budget.ts", "script:diff-nango-shadow": "tsx src/bin/diff-nango-shadow.ts", - "script:diff-shadow-unit-detail": "tsx src/bin/diff-shadow-unit-detail.ts" + "script:diff-shadow-unit-detail": "tsx src/bin/diff-shadow-unit-detail.ts", + "script:backfill-octolens-mentions": "tsx src/bin/backfill-octolens-mentions.ts" }, "dependencies": { "@crowd/archetype-standard": "workspace:*", diff --git a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts new file mode 100644 index 0000000000..61a60355ad --- /dev/null +++ b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts @@ -0,0 +1,151 @@ +import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' +import { + listActiveOctolensKeywordMappings, + listOctolensAnchorIntegrationIds, + listOctolensMentionPollUnits, + queueOctolensKeywordBackfills, + rewindOctolensMentionPollStartDate, +} from '@crowd/data-access-layer/src/octolens' +import type { + IOctolensMentionPollUnit, + IOctolensMentionPollWriteResult, +} from '@crowd/data-access-layer/src/octolens' +import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' +import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' +import { getServiceLogger } from '@crowd/logging' +import { TEMPORAL_CONFIG, WorkflowNotFoundError, getTemporalClient } from '@crowd/temporal' +import { PlatformType } from '@crowd/types' + +import { + OCTOLENS_BACKFILL_USAGE, + parseOctolensBackfillArgs, + planOctolensBackfill, +} from '../octolensBackfill' +import type { OctolensBackfillArgs, OctolensBackfillPlan } from '../octolensBackfill' +import { syncRunWorkflowId } from '../syncRunWorkflowId' + +const log = getServiceLogger() + +const RETRY_HINT = 'retry in a few minutes, once the current mention poll run has finished' + +function parseArgsOrExit(): OctolensBackfillArgs { + try { + return parseOctolensBackfillArgs(process.argv.slice(2), new Date()) + } catch (err) { + log.error((err as Error).message) + log.error(OCTOLENS_BACKFILL_USAGE) + process.exit(1) + } +} + +async function loadPlan(qx: QueryExecutor, args: OctolensBackfillArgs) { + const [anchorIntegrationIds, units, mappings] = await Promise.all([ + listOctolensAnchorIntegrationIds(qx), + listOctolensMentionPollUnits(qx), + listActiveOctolensKeywordMappings(qx), + ]) + return planOctolensBackfill(args, { + anchorIntegrationIds, + units, + activeKeywordIds: new Set(mappings.map((m) => m.keywordId)), + }) +} + +// The dispatcher clears lockedAt when it re-claims a unit whose run outlived the claim lease, +// so only Temporal knows a long run is still going and will overwrite the watermark at its end. +async function isRunInFlight(unit: IOctolensMentionPollUnit): Promise { + const temporal = await getTemporalClient(TEMPORAL_CONFIG()) + const workflowId = syncRunWorkflowId({ + id: unit.id, + platform: PlatformType.OCTOLENS, + syncName: 'mentionPoll', + channelName: unit.channelName, + }) + try { + const description = await temporal.workflow.getHandle(workflowId).describe() + return description.status.name === 'RUNNING' + } catch (err) { + if (err instanceof WorkflowNotFoundError) return false + throw err + } +} + +function describePlan(plan: OctolensBackfillPlan): Record { + const base = { unitId: plan.unit.id, integrationId: plan.unit.integrationId } + if (plan.mode === 'keywords') { + return { + ...base, + keywordIdsToQueue: plan.keywordIdsToQueue, + alreadyQueued: plan.alreadyQueued, + currentBackfill: plan.unit.watermark?.backfill ?? [], + } + } + return { + ...base, + previousStartDate: plan.previousStartDate, + startDate: plan.startDate, + droppedCursor: plan.unit.watermark?.cursor ?? null, + droppedPendingMaxTimestamp: plan.unit.watermark?.pendingMaxTimestamp ?? null, + } +} + +function applyPlan( + qx: QueryExecutor, + plan: OctolensBackfillPlan, +): Promise { + if (plan.mode === 'keywords') { + return queueOctolensKeywordBackfills(qx, plan.unit.id, plan.keywordIdsToQueue) + } + return rewindOctolensMentionPollStartDate(qx, plan.unit.id, plan.startDate) +} + +setImmediate(async () => { + const args = parseArgsOrExit() + try { + const qx = pgpQx(await getDbConnection(WRITE_DB_CONFIG())) + const plan = await loadPlan(qx, args) + const summary = describePlan(plan) + + if (plan.mode === 'keywords' && plan.keywordIdsToQueue.length === 0) { + log.info(summary, 'nothing to do: every keyword is already queued for backfill') + process.exit(0) + } + + const busy = plan.unit.lockedAt !== null || (await isRunInFlight(plan.unit)) + + if (args.dryRun) { + log.info({ ...summary, busy }, 'dry run: would update the octolens mention poll watermark') + if (busy) { + log.warn(`a mention poll run is in progress, a real run would refuse now: ${RETRY_HINT}`) + } + process.exit(0) + } + + if (busy) { + log.error(summary, `refused: a mention poll run is in progress; ${RETRY_HINT}`) + process.exit(2) + } + + const result = await applyPlan(qx, plan) + if (result.outcome === 'locked') { + log.error(summary, `refused: a mention poll run claimed the unit just now; ${RETRY_HINT}`) + process.exit(2) + } + if (result.outcome !== 'applied') { + log.error( + { ...summary, outcome: result.outcome }, + 'refused: the sync unit is no longer active', + ) + process.exit(1) + } + + log.info( + { ...summary, watermark: result.watermark }, + 'octolens mention poll watermark updated; the next poll runs pick it up', + ) + process.exit(0) + } catch (err) { + log.error(err, 'octolens mention backfill failed') + process.exit(1) + } +}) diff --git a/services/apps/connectors_worker/src/octolensBackfill.test.ts b/services/apps/connectors_worker/src/octolensBackfill.test.ts new file mode 100644 index 0000000000..774433c9f2 --- /dev/null +++ b/services/apps/connectors_worker/src/octolensBackfill.test.ts @@ -0,0 +1,133 @@ +import { describe, expect, it } from 'vitest' + +import type { IOctolensMentionPollUnit } from '@crowd/data-access-layer/src/octolens' + +import { parseOctolensBackfillArgs, planOctolensBackfill } from './octolensBackfill' +import type { IOctolensBackfillState, OctolensBackfillArgs } from './octolensBackfill' + +const NOW = new Date('2026-10-16T12:00:00.000Z') + +function parse(...argv: string[]): OctolensBackfillArgs { + return parseOctolensBackfillArgs(argv, NOW) +} + +function unit(overrides: Partial = {}): IOctolensMentionPollUnit { + return { + id: 'unit-1', + integrationId: 'anchor-1', + channelName: 'octolens-mentions', + status: 'active', + lockedAt: null, + watermark: { + startDate: '2026-10-16T11:55:00.000Z', + cursor: null, + pendingMaxTimestamp: null, + backfill: [{ keywordId: 7, cursor: 'k7-page2' }], + }, + ...overrides, + } +} + +function state(overrides: Partial = {}): IOctolensBackfillState { + return { + anchorIntegrationIds: ['anchor-1'], + units: [unit()], + activeKeywordIds: new Set([7, 8, 9]), + ...overrides, + } +} + +describe('parseOctolensBackfillArgs', () => { + it('parses keyword ids, dropping duplicates and the pnpm separator', () => { + expect(parse('--', '--keywords', '9,8, 9', '--dry-run')).toEqual({ + mode: 'keywords', + keywordIds: [9, 8], + dryRun: true, + }) + }) + + it('normalizes --from to an ISO timestamp', () => { + expect(parse('--from', '2026-09-01')).toEqual({ + mode: 'from', + startDate: '2026-09-01T00:00:00.000Z', + dryRun: false, + }) + expect(parse('--from', '2026-09-01T10:30:00+02:00')).toMatchObject({ + startDate: '2026-09-01T08:30:00.000Z', + }) + }) + + it.each([ + [[], 'exactly one'], + [['--keywords', '1', '--from', '2026-09-01'], 'exactly one'], + [['--keywords'], 'needs a value'], + [['--keywords', '--dry-run'], 'needs a value'], + [['--keywords', '1', '--keywords', '2'], 'more than once'], + [['--keywords', '1', '--dry-run', '--dry-run'], 'more than once'], + [['--keywords', '1', '--force'], 'unknown argument'], + [['--keywords', '0'], 'positive integer'], + [['--keywords', '-3'], 'positive integer'], + [['--keywords', '1.5'], 'positive integer'], + [['--keywords', '1,,2'], 'positive integer'], + [['--keywords', 'abc'], 'positive integer'], + [['--keywords', '2147483648'], 'positive integer'], + [['--from', 'yesterday'], 'expected YYYY-MM-DD'], + [['--from', '2026-09-01T10:00:00'], 'expected YYYY-MM-DD'], + [['--from', '2026-02-30'], 'not a real date'], + [['--from', '2026-13-01'], 'not a real date'], + [['--from', '2026-10-17'], 'in the future'], + ])('rejects %j', (argv, message) => { + expect(() => parse(...argv)).toThrow(message) + }) +}) + +describe('planOctolensBackfill', () => { + it('queues only keywords that are not already in the backfill queue', () => { + const plan = planOctolensBackfill(parse('--keywords', '7,9'), state()) + + expect(plan).toMatchObject({ mode: 'keywords', keywordIdsToQueue: [9], alreadyQueued: [7] }) + }) + + it('refuses keywords without an active subproject mapping', () => { + expect(() => planOctolensBackfill(parse('--keywords', '9,42,43'), state())).toThrow( + 'keywords without an active subproject mapping: 42, 43', + ) + }) + + it('rewinds to a date earlier than the current start date', () => { + const plan = planOctolensBackfill(parse('--from', '2026-09-01'), state()) + + expect(plan).toMatchObject({ + mode: 'from', + startDate: '2026-09-01T00:00:00.000Z', + previousStartDate: '2026-10-16T11:55:00.000Z', + }) + }) + + it('refuses a --from that would move the start date forward', () => { + const current = unit({ watermark: { startDate: '2026-09-01T00:00:00.000Z' } }) + + expect(() => + planOctolensBackfill(parse('--from', '2026-09-02'), state({ units: [current] })), + ).toThrow('not earlier than the current start date') + }) + + it('accepts any past --from on a unit that has never run', () => { + const fresh = unit({ watermark: null }) + + expect( + planOctolensBackfill(parse('--from', '2026-09-02'), state({ units: [fresh] })), + ).toMatchObject({ previousStartDate: null }) + }) + + it.each([ + [{ anchorIntegrationIds: [] }, 'exactly one live segment-less octolens integration, found 0'], + [{ anchorIntegrationIds: ['a', 'b'] }, 'found 2'], + [{ units: [] }, 'exactly one octolens mentionPoll sync unit, found 0'], + [{ units: [unit(), unit({ id: 'unit-2' })] }, 'found 2'], + [{ units: [unit({ status: 'dead_letter' })] }, 'is dead_letter, not active'], + [{ units: [unit({ status: 'paused' })] }, 'is paused, not active'], + ])('refuses state %j', (overrides, message) => { + expect(() => planOctolensBackfill(parse('--keywords', '9'), state(overrides))).toThrow(message) + }) +}) diff --git a/services/apps/connectors_worker/src/octolensBackfill.ts b/services/apps/connectors_worker/src/octolensBackfill.ts new file mode 100644 index 0000000000..09a7ab723c --- /dev/null +++ b/services/apps/connectors_worker/src/octolensBackfill.ts @@ -0,0 +1,156 @@ +import type { IOctolensMentionPollUnit } from '@crowd/data-access-layer/src/octolens' + +const MAX_POSTGRES_INT = 2_147_483_647 +const ISO_DATE = + /^(\d{4})-(\d{2})-(\d{2})(?:T\d{2}:\d{2}(?::\d{2}(?:\.\d{1,3})?)?(?:Z|[+-]\d{2}:\d{2}))?$/ + +export type OctolensBackfillArgs = + | { mode: 'keywords'; keywordIds: number[]; dryRun: boolean } + | { mode: 'from'; startDate: string; dryRun: boolean } + +export type OctolensBackfillPlan = + | { + mode: 'keywords' + unit: IOctolensMentionPollUnit + keywordIdsToQueue: number[] + alreadyQueued: number[] + } + | { + mode: 'from' + unit: IOctolensMentionPollUnit + startDate: string + previousStartDate: string | null + } + +export interface IOctolensBackfillState { + anchorIntegrationIds: string[] + units: IOctolensMentionPollUnit[] + activeKeywordIds: Set +} + +export const OCTOLENS_BACKFILL_USAGE = + 'Usage: backfill-octolens-mentions (--keywords | --from ) [--dry-run]' + +export function parseOctolensBackfillArgs(argv: string[], now: Date): OctolensBackfillArgs { + const values = new Map() + let dryRun = false + + const args = argv.filter((arg) => arg !== '--') + for (let i = 0; i < args.length; i++) { + const arg = args[i] + if (arg === '--dry-run') { + if (dryRun) throw new Error('--dry-run given more than once') + dryRun = true + continue + } + if (arg !== '--keywords' && arg !== '--from') { + throw new Error(`unknown argument "${arg}"`) + } + if (values.has(arg)) throw new Error(`${arg} given more than once`) + const value = args[i + 1] + if (value === undefined || value.startsWith('--')) throw new Error(`${arg} needs a value`) + values.set(arg, value) + i++ + } + + const keywords = values.get('--keywords') + const from = values.get('--from') + if ((keywords === undefined) === (from === undefined)) { + throw new Error('pass exactly one of --keywords or --from') + } + + if (keywords !== undefined) { + return { mode: 'keywords', keywordIds: parseKeywordIds(keywords), dryRun } + } + return { mode: 'from', startDate: parseStartDate(from as string, now), dryRun } +} + +function parseKeywordIds(raw: string): number[] { + const ids = raw.split(',').map((part) => { + const trimmed = part.trim() + const id = Number(trimmed) + if (!/^\d+$/.test(trimmed) || id < 1 || id > MAX_POSTGRES_INT) { + throw new Error(`invalid keyword id "${part}": expected a positive integer`) + } + return id + }) + return [...new Set(ids)] +} + +function parseStartDate(raw: string, now: Date): string { + const match = ISO_DATE.exec(raw) + if (!match) { + throw new Error(`invalid --from "${raw}": expected YYYY-MM-DD or an ISO timestamp with a zone`) + } + const [year, month, day] = match.slice(1, 4).map(Number) + const calendarDay = new Date(Date.UTC(year, month - 1, day)) + const date = new Date(raw) + if ( + calendarDay.getUTCFullYear() !== year || + calendarDay.getUTCMonth() !== month - 1 || + calendarDay.getUTCDate() !== day || + Number.isNaN(date.getTime()) + ) { + throw new Error(`invalid --from "${raw}": not a real date`) + } + if (date.getTime() > now.getTime()) { + throw new Error(`invalid --from "${raw}": in the future`) + } + return date.toISOString() +} + +export function planOctolensBackfill( + args: OctolensBackfillArgs, + state: IOctolensBackfillState, +): OctolensBackfillPlan { + if (state.anchorIntegrationIds.length !== 1) { + throw new Error( + `expected exactly one live segment-less octolens integration, found ${state.anchorIntegrationIds.length}`, + ) + } + if (state.units.length !== 1) { + throw new Error( + `expected exactly one octolens mentionPoll sync unit, found ${state.units.length}`, + ) + } + const unit = state.units[0] + if (unit.status !== 'active') { + throw new Error( + `octolens mentionPoll sync unit ${unit.id} is ${unit.status}, not active: fix the unit first`, + ) + } + + if (args.mode === 'keywords') { + const unmapped = args.keywordIds.filter((id) => !state.activeKeywordIds.has(id)) + if (unmapped.length > 0) { + throw new Error(`keywords without an active subproject mapping: ${unmapped.join(', ')}`) + } + const queued = new Set(queuedKeywordIds(unit.watermark)) + return { + mode: 'keywords', + unit, + keywordIdsToQueue: args.keywordIds.filter((id) => !queued.has(id)), + alreadyQueued: args.keywordIds.filter((id) => queued.has(id)), + } + } + + const previousStartDate = + typeof unit.watermark?.startDate === 'string' ? unit.watermark.startDate : null + if (previousStartDate !== null) { + const previous = new Date(previousStartDate).getTime() + if (!Number.isNaN(previous) && new Date(args.startDate).getTime() >= previous) { + throw new Error( + `--from ${args.startDate} is not earlier than the current start date ${previousStartDate}`, + ) + } + } + return { mode: 'from', unit, startDate: args.startDate, previousStartDate } +} + +function queuedKeywordIds(watermark: Record | null): number[] { + const backfill = watermark?.backfill + if (!Array.isArray(backfill)) return [] + return backfill + .map((item) => (item as { keywordId?: unknown } | null)?.keywordId) + .filter((id): id is number => Number.isInteger(id)) +} From 82ac9805cba5095b56502c629cefab8676900e65 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:37:24 +0200 Subject: [PATCH 04/13] feat: cap the regular octolens mention poll at 20 pages per run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../octolens/syncs/mentionPoll.test.ts | 66 +++++++++++++++++++ .../connectors/octolens/syncs/mentionPoll.ts | 15 ++++- 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts index 64cfc9654e..6fcc2fceaa 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts @@ -729,3 +729,69 @@ describe('runMentionPoll keyword backfill (queued manually)', () => { expect(watermarks[watermarks.length - 1]).toMatchObject({ backfill: [] }) }) }) + +describe('runMentionPoll page cap', () => { + const START = '2026-10-01T00:00:00.000Z' + const QUEUED = [{ keywordId: 7, cursor: null }] + + it('stops the regular poll at 20 pages and keeps the cycle, without starting the backfill', async () => { + for (let i = 1; i <= 25; i++) { + mockedFetchMentionsPage.mockResolvedValueOnce(page([mention()], `cursor-${i}`)) + } + const { ctx, watermarks } = makeCtx({ watermark: { startDate: START, backfill: QUEUED } }) + + expect(await runMentionPoll(ctx)).toEqual({ complete: true }) + expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(20) + expect(mockedFetchMentionsPage).not.toHaveBeenCalledWith('test-token', { + keywordIds: [7], + cursor: undefined, + }) + expect(watermarks[watermarks.length - 1]).toEqual({ + startDate: START, + cursor: 'cursor-20', + pendingMaxTimestamp: '2026-10-12T08:00:00.000Z', + backfill: QUEUED, + }) + }) + + it('resumes the capped cycle from its cursor on the next run, then runs the backfill', async () => { + mockedFetchMentionsPage + .mockResolvedValueOnce(page([mention()], null)) + .mockResolvedValueOnce(page([])) + const { ctx, watermarks } = makeCtx({ + watermark: { + startDate: START, + cursor: 'cursor-20', + pendingMaxTimestamp: '2026-10-12T08:00:00.000Z', + backfill: QUEUED, + }, + }) + + expect(await runMentionPoll(ctx)).toEqual({ complete: true }) + expect(mockedFetchMentionsPage.mock.calls.map(([, params]) => params)).toEqual([ + { startDate: START, cursor: 'cursor-20' }, + { keywordIds: [7], cursor: undefined }, + ]) + expect(watermarks[watermarks.length - 1]).toEqual({ + startDate: '2026-10-12T08:00:00.000Z', + cursor: null, + pendingMaxTimestamp: null, + backfill: [], + }) + }) + + it('finishes a cycle that drains exactly on the 20th page', async () => { + for (let i = 1; i < 20; i++) { + mockedFetchMentionsPage.mockResolvedValueOnce(page([], `cursor-${i}`)) + } + mockedFetchMentionsPage.mockResolvedValueOnce(page([mention()], null)) + const { ctx, watermarks } = makeCtx({ watermark: { startDate: START } }) + + expect(await runMentionPoll(ctx)).toEqual({ complete: true }) + expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(20) + expect(watermarks[watermarks.length - 1]).toMatchObject({ + startDate: '2026-10-12T08:00:00.000Z', + cursor: null, + }) + }) +}) diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts index c5b6e61831..de27799550 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts @@ -6,8 +6,9 @@ import type { MappedOctolensMention } from '../mapper' import { fetchMentionsPage } from '../mentionsClient' import type { OctolensMention } from '../mentionsClient' -// Sized for the worst case, every page retried (4 attempts), at the 5-minute cadence: 8 × 4 × 12 = -// 384 requests/hour, leaving the shared 500/hour budget room for the regular poll and discovery. +// Worst case per run is 20 + 8 = 28 pages, 336/hour at the 5-minute cadence before retries, against +// Octolens's shared 500/hour budget; a regular cycle longer than 20 pages resumes next cadence. +const MAX_POLL_PAGES_PER_RUN = 20 const MAX_BACKFILL_PAGES_PER_RUN = 8 interface KeywordBackfill { @@ -148,7 +149,7 @@ export async function runMentionPoll(ctx: SyncContext): Promise): Promise= MAX_POLL_PAGES_PER_RUN) { + // complete: true schedules the next run at the normal cadence; false would retry within ~1 minute. + ctx.log.info( + { pagesFetched }, + 'octolens mention poll paused at the page cap until next cadence', + ) + return { complete: true } + } if (!polled) { ctx.log.info({ pagesFetched }, 'octolens mention poll deferred: run budget exhausted') return { complete: false } From 68077cadefb1170b59c5748ab3029e19f29faa66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:37:26 +0200 Subject: [PATCH 05/13] fix: refuse ambiguous keywords and old rewinds in the octolens backfill script MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/bin/backfill-octolens-mentions.ts | 24 ++++-------- .../src/octolensBackfill.test.ts | 35 +++++++++++++++--- .../connectors_worker/src/octolensBackfill.ts | 34 ++++++++++------- .../data-access-layer/src/octolens/index.ts | 20 ++-------- .../src/octolens/mentionPollWatermark.test.ts | 37 +++++++++++++++++-- 5 files changed, 95 insertions(+), 55 deletions(-) diff --git a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts index 61a60355ad..094b49a6bd 100644 --- a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts +++ b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts @@ -1,7 +1,6 @@ import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' import { listActiveOctolensKeywordMappings, - listOctolensAnchorIntegrationIds, listOctolensMentionPollUnits, queueOctolensKeywordBackfills, rewindOctolensMentionPollStartDate, @@ -14,7 +13,6 @@ import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' import { getServiceLogger } from '@crowd/logging' import { TEMPORAL_CONFIG, WorkflowNotFoundError, getTemporalClient } from '@crowd/temporal' -import { PlatformType } from '@crowd/types' import { OCTOLENS_BACKFILL_USAGE, @@ -39,28 +37,22 @@ function parseArgsOrExit(): OctolensBackfillArgs { } async function loadPlan(qx: QueryExecutor, args: OctolensBackfillArgs) { - const [anchorIntegrationIds, units, mappings] = await Promise.all([ - listOctolensAnchorIntegrationIds(qx), + const [units, mappings] = await Promise.all([ listOctolensMentionPollUnits(qx), listActiveOctolensKeywordMappings(qx), ]) - return planOctolensBackfill(args, { - anchorIntegrationIds, - units, - activeKeywordIds: new Set(mappings.map((m) => m.keywordId)), - }) + const activeMappingCounts = new Map() + for (const { keywordId } of mappings) { + activeMappingCounts.set(keywordId, (activeMappingCounts.get(keywordId) ?? 0) + 1) + } + return planOctolensBackfill(args, { units, activeMappingCounts }) } // The dispatcher clears lockedAt when it re-claims a unit whose run outlived the claim lease, // so only Temporal knows a long run is still going and will overwrite the watermark at its end. async function isRunInFlight(unit: IOctolensMentionPollUnit): Promise { const temporal = await getTemporalClient(TEMPORAL_CONFIG()) - const workflowId = syncRunWorkflowId({ - id: unit.id, - platform: PlatformType.OCTOLENS, - syncName: 'mentionPoll', - channelName: unit.channelName, - }) + const workflowId = syncRunWorkflowId(unit) try { const description = await temporal.workflow.getHandle(workflowId).describe() return description.status.name === 'RUNNING' @@ -134,7 +126,7 @@ setImmediate(async () => { if (result.outcome !== 'applied') { log.error( { ...summary, outcome: result.outcome }, - 'refused: the sync unit is no longer active', + 'refused: the sync unit is no longer active or was removed', ) process.exit(1) } diff --git a/services/apps/connectors_worker/src/octolensBackfill.test.ts b/services/apps/connectors_worker/src/octolensBackfill.test.ts index 774433c9f2..a0ad57bade 100644 --- a/services/apps/connectors_worker/src/octolensBackfill.test.ts +++ b/services/apps/connectors_worker/src/octolensBackfill.test.ts @@ -15,6 +15,8 @@ function unit(overrides: Partial = {}): IOctolensMenti return { id: 'unit-1', integrationId: 'anchor-1', + platform: 'octolens', + syncName: 'mentionPoll', channelName: 'octolens-mentions', status: 'active', lockedAt: null, @@ -30,18 +32,21 @@ function unit(overrides: Partial = {}): IOctolensMenti function state(overrides: Partial = {}): IOctolensBackfillState { return { - anchorIntegrationIds: ['anchor-1'], units: [unit()], - activeKeywordIds: new Set([7, 8, 9]), + activeMappingCounts: new Map([ + [7, 1], + [8, 1], + [9, 1], + ]), ...overrides, } } describe('parseOctolensBackfillArgs', () => { - it('parses keyword ids, dropping duplicates and the pnpm separator', () => { + it('parses keyword ids, dropping the pnpm separator', () => { expect(parse('--', '--keywords', '9,8, 9', '--dry-run')).toEqual({ mode: 'keywords', - keywordIds: [9, 8], + keywordIds: [9, 8, 9], dryRun: true, }) }) @@ -76,6 +81,9 @@ describe('parseOctolensBackfillArgs', () => { [['--from', '2026-02-30'], 'not a real date'], [['--from', '2026-13-01'], 'not a real date'], [['--from', '2026-10-17'], 'in the future'], + [['--from', '2025-10-16T11:59:59Z'], 'more than 1 year back'], + [['--from', '2026-09-01T24:00Z'], 'expected YYYY-MM-DD'], + [['--from', '2026-09-01T10:60Z'], 'expected YYYY-MM-DD'], ])('rejects %j', (argv, message) => { expect(() => parse(...argv)).toThrow(message) }) @@ -88,6 +96,23 @@ describe('planOctolensBackfill', () => { expect(plan).toMatchObject({ mode: 'keywords', keywordIdsToQueue: [9], alreadyQueued: [7] }) }) + it('accepts --from exactly one year back', () => { + expect(parse('--from', '2025-10-16T12:00:00Z')).toMatchObject({ + startDate: '2025-10-16T12:00:00.000Z', + }) + }) + + it('refuses a keyword mapped to more than one subproject', () => { + const counts = new Map([ + [9, 1], + [8, 2], + ]) + + expect(() => + planOctolensBackfill(parse('--keywords', '9,8'), state({ activeMappingCounts: counts })), + ).toThrow('keyword 8 is mapped to more than one subproject') + }) + it('refuses keywords without an active subproject mapping', () => { expect(() => planOctolensBackfill(parse('--keywords', '9,42,43'), state())).toThrow( 'keywords without an active subproject mapping: 42, 43', @@ -121,8 +146,6 @@ describe('planOctolensBackfill', () => { }) it.each([ - [{ anchorIntegrationIds: [] }, 'exactly one live segment-less octolens integration, found 0'], - [{ anchorIntegrationIds: ['a', 'b'] }, 'found 2'], [{ units: [] }, 'exactly one octolens mentionPoll sync unit, found 0'], [{ units: [unit(), unit({ id: 'unit-2' })] }, 'found 2'], [{ units: [unit({ status: 'dead_letter' })] }, 'is dead_letter, not active'], diff --git a/services/apps/connectors_worker/src/octolensBackfill.ts b/services/apps/connectors_worker/src/octolensBackfill.ts index 09a7ab723c..b926dbd349 100644 --- a/services/apps/connectors_worker/src/octolensBackfill.ts +++ b/services/apps/connectors_worker/src/octolensBackfill.ts @@ -1,8 +1,9 @@ import type { IOctolensMentionPollUnit } from '@crowd/data-access-layer/src/octolens' const MAX_POSTGRES_INT = 2_147_483_647 +const MAX_REWIND_YEARS = 1 const ISO_DATE = - /^(\d{4})-(\d{2})-(\d{2})(?:T\d{2}:\d{2}(?::\d{2}(?:\.\d{1,3})?)?(?:Z|[+-]\d{2}:\d{2}))?$/ + /^(\d{4})-(\d{2})-(\d{2})(?:T(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d{1,3})?)?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d))?$/ export type OctolensBackfillArgs = | { mode: 'keywords'; keywordIds: number[]; dryRun: boolean } @@ -23,13 +24,15 @@ export type OctolensBackfillPlan = } export interface IOctolensBackfillState { - anchorIntegrationIds: string[] units: IOctolensMentionPollUnit[] - activeKeywordIds: Set + activeMappingCounts: Map } -export const OCTOLENS_BACKFILL_USAGE = - 'Usage: backfill-octolens-mentions (--keywords | --from ) [--dry-run]' +export const OCTOLENS_BACKFILL_USAGE = [ + 'Usage: backfill-octolens-mentions (--keywords | --from ) [--dry-run]', + ' --from takes YYYY-MM-DD (UTC midnight) or an ISO timestamp with a zone, at most 1 year back.', + ' Exit codes: 0 done or nothing to do, 1 invalid input or state, 2 a poll run is in progress (retry).', +].join('\n') export function parseOctolensBackfillArgs(argv: string[], now: Date): OctolensBackfillArgs { const values = new Map() @@ -66,7 +69,7 @@ export function parseOctolensBackfillArgs(argv: string[], now: Date): OctolensBa } function parseKeywordIds(raw: string): number[] { - const ids = raw.split(',').map((part) => { + return raw.split(',').map((part) => { const trimmed = part.trim() const id = Number(trimmed) if (!/^\d+$/.test(trimmed) || id < 1 || id > MAX_POSTGRES_INT) { @@ -74,7 +77,6 @@ function parseKeywordIds(raw: string): number[] { } return id }) - return [...new Set(ids)] } function parseStartDate(raw: string, now: Date): string { @@ -96,6 +98,13 @@ function parseStartDate(raw: string, now: Date): string { if (date.getTime() > now.getTime()) { throw new Error(`invalid --from "${raw}": in the future`) } + const earliest = new Date(now) + earliest.setUTCFullYear(earliest.getUTCFullYear() - MAX_REWIND_YEARS) + if (date.getTime() < earliest.getTime()) { + throw new Error( + `invalid --from "${raw}": more than ${MAX_REWIND_YEARS} year back (earliest ${earliest.toISOString()})`, + ) + } return date.toISOString() } @@ -103,11 +112,6 @@ export function planOctolensBackfill( args: OctolensBackfillArgs, state: IOctolensBackfillState, ): OctolensBackfillPlan { - if (state.anchorIntegrationIds.length !== 1) { - throw new Error( - `expected exactly one live segment-less octolens integration, found ${state.anchorIntegrationIds.length}`, - ) - } if (state.units.length !== 1) { throw new Error( `expected exactly one octolens mentionPoll sync unit, found ${state.units.length}`, @@ -121,10 +125,14 @@ export function planOctolensBackfill( } if (args.mode === 'keywords') { - const unmapped = args.keywordIds.filter((id) => !state.activeKeywordIds.has(id)) + const unmapped = args.keywordIds.filter((id) => !state.activeMappingCounts.has(id)) if (unmapped.length > 0) { throw new Error(`keywords without an active subproject mapping: ${unmapped.join(', ')}`) } + const ambiguous = args.keywordIds.find((id) => (state.activeMappingCounts.get(id) ?? 0) > 1) + if (ambiguous !== undefined) { + throw new Error(`keyword ${ambiguous} is mapped to more than one subproject`) + } const queued = new Set(queuedKeywordIds(unit.watermark)) return { mode: 'keywords', diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts index 1ed15f9b2f..ff4448cdde 100644 --- a/services/libs/data-access-layer/src/octolens/index.ts +++ b/services/libs/data-access-layer/src/octolens/index.ts @@ -279,6 +279,8 @@ const MENTION_POLL_SYNC_NAME = 'mentionPoll' export interface IOctolensMentionPollUnit { id: string integrationId: string + platform: string + syncName: string channelName: string status: string lockedAt: Date | null @@ -292,27 +294,13 @@ export interface IOctolensMentionPollWriteResult { watermark: Record | null } -export async function listOctolensAnchorIntegrationIds(qx: QueryExecutor): Promise { - const rows: { id: string }[] = await qx.select( - ` - SELECT i.id - FROM public.integrations i - WHERE i.platform = $(platform) - AND i."segmentId" IS NULL - AND i."deletedAt" IS NULL - ORDER BY i.id - `, - { platform: PlatformType.OCTOLENS }, - ) - return rows.map((r) => r.id) -} - export async function listOctolensMentionPollUnits( qx: QueryExecutor, ): Promise { return qx.select( ` - SELECT su.id, su."integrationId", su."channelName", su.status, su."lockedAt", su.watermark + SELECT su.id, su."integrationId", su.platform, su."syncName", su."channelName", su.status, + su."lockedAt", su.watermark FROM integration.sync_units su JOIN public.integrations i ON i.id = su."integrationId" WHERE su.platform = $(platform) diff --git a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts index 827005b2b2..782a808fc2 100644 --- a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts +++ b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts @@ -5,7 +5,6 @@ import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' import { - listOctolensAnchorIntegrationIds, listOctolensMentionPollUnits, queueOctolensKeywordBackfills, rewindOctolensMentionPollStartDate, @@ -81,10 +80,9 @@ describe('listOctolensMentionPollUnits', () => { const deletedAnchorId = await createAnchorIntegration(qx, { deletedAt: '2026-01-01' }) await createMentionPollUnit(qx, deletedAnchorId) - expect(await listOctolensAnchorIntegrationIds(qx)).toEqual([anchorId]) const units = await listOctolensMentionPollUnits(qx) - expect(units.map((u) => [u.id, u.integrationId, u.status])).toEqual([ - [unitId, anchorId, 'dead_letter'], + expect(units.map((u) => [u.id, u.integrationId, u.platform, u.syncName, u.status])).toEqual([ + [unitId, anchorId, 'octolens', 'mentionPoll', 'dead_letter'], ]) }) }) @@ -118,6 +116,37 @@ describe('queueOctolensKeywordBackfills', () => { expect(await readWatermark(qx, unitId)).toEqual({ backfill: [{ keywordId: 5, cursor: null }] }) }) + test('adds the queue to a stored watermark without a backfill key', async ({ qx }) => { + const poll = { + startDate: MID_CYCLE.startDate, + cursor: MID_CYCLE.cursor, + pendingMaxTimestamp: MID_CYCLE.pendingMaxTimestamp, + } + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + watermark: poll, + }) + + await queueOctolensKeywordBackfills(qx, unitId, [5]) + + expect(await readWatermark(qx, unitId)).toEqual({ + ...poll, + backfill: [{ keywordId: 5, cursor: null }], + }) + }) + + test('replaces a backfill value that is not an array', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + watermark: { ...MID_CYCLE, backfill: { keywordId: 7 } }, + }) + + await queueOctolensKeywordBackfills(qx, unitId, [5]) + + expect(await readWatermark(qx, unitId)).toEqual({ + ...MID_CYCLE, + backfill: [{ keywordId: 5, cursor: null }], + }) + }) + test('refuses and writes nothing while the unit is claimed', async ({ qx }) => { const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { locked: true, From 6f77030f581fe80b25acea30f157e4d1b110cfcf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:50:31 +0200 Subject: [PATCH 06/13] fix: keep a still-running sync unit claimed instead of clearing its lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../activities/dispatcherActivities.test.ts | 72 ++++++++++++++++++ .../src/activities/dispatcherActivities.ts | 13 +++- .../src/connectors/syncUnits.test.ts | 75 ++++++++++++++++++- .../src/connectors/syncUnits.ts | 15 ++++ 4 files changed, 171 insertions(+), 4 deletions(-) create mode 100644 services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts new file mode 100644 index 0000000000..e1cb8613d5 --- /dev/null +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts @@ -0,0 +1,72 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { startRun } from './dispatcherActivities' + +const PROBE_AT = new Date('2026-10-16T12:05:00.000Z') + +const mocks = vi.hoisted(() => ({ + start: vi.fn(), + describe: vi.fn(), + renewRunningUnitLease: vi.fn(), + rescheduleUnit: vi.fn(), +})) + +vi.mock('../main', () => ({ + svc: { + postgres: { writer: {} }, + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + temporal: { + workflow: { start: mocks.start, getHandle: () => ({ describe: mocks.describe }) }, + }, + }, +})) + +vi.mock('@crowd/data-access-layer/src/connectors', () => ({ + claimDueUnits: vi.fn(), + renewRunningUnitLease: mocks.renewRunningUnitLease, + rescheduleUnit: mocks.rescheduleUnit, +})) + +vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({ + dbStoreQx: vi.fn(() => ({})), +})) + +vi.mock('../scheduling', () => ({ + runningProbeRunAt: () => PROBE_AT, + shortDeferRunAt: () => PROBE_AT, +})) + +const UNIT = { + id: 'unit-1', + integrationId: 'integration-1', + platform: 'github', + syncName: 'issues', + channelId: 'repo', + channelName: 'https://github.com/org/repo', +} + +beforeEach(() => { + vi.clearAllMocks() + mocks.describe.mockResolvedValue({ startTime: new Date('2026-10-16T11:00:00.000Z') }) + mocks.renewRunningUnitLease.mockResolvedValue(1) +}) + +describe('startRun', () => { + it('starts a workflow without touching the claim', async () => { + mocks.start.mockResolvedValue(undefined) + + expect(await startRun(UNIT)).toBe('started') + expect(mocks.renewRunningUnitLease).not.toHaveBeenCalled() + expect(mocks.rescheduleUnit).not.toHaveBeenCalled() + }) + + it('keeps a still-running unit claimed instead of releasing it', async () => { + const alreadyStarted = new Error('already started') + alreadyStarted.name = 'WorkflowExecutionAlreadyStartedError' + mocks.start.mockRejectedValue(alreadyStarted) + + expect(await startRun(UNIT)).toBe('alreadyRunning') + expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id, PROBE_AT) + expect(mocks.rescheduleUnit).not.toHaveBeenCalled() + }) +}) diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts index 6052e1fae5..507d58b4d1 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -1,5 +1,9 @@ import { createTokenPool, findManifest, mapWithConcurrency } from '@crowd/connectors' -import { claimDueUnits, rescheduleUnit } from '@crowd/data-access-layer/src/connectors' +import { + claimDueUnits, + renewRunningUnitLease, + rescheduleUnit, +} from '@crowd/data-access-layer/src/connectors' import type { IClaimedUnit } from '@crowd/data-access-layer/src/connectors' import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor' import { RedisCache } from '@crowd/redis' @@ -76,8 +80,11 @@ async function backOffRunningUnit(unit: IClaimedUnit): Promise { .then(runningProbeRunAt) .catch(() => shortDeferRunAt()) - await rescheduleUnit(dbStoreQx(svc.postgres.writer), unit.id, nextRunAt) - svc.log.info({ unitId: unit.id, nextRunAt }, 'unit already running, rescheduled probe') + const renewed = await renewRunningUnitLease(dbStoreQx(svc.postgres.writer), unit.id, nextRunAt) + svc.log.info( + { unitId: unit.id, nextRunAt, renewed: renewed > 0 }, + 'unit already running, lease renewed until the next probe', + ) } async function runningSince(unit: IClaimedUnit): Promise { 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 655797dca7..a69080b6d6 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -4,7 +4,13 @@ import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common' import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' -import { listShadowDiffUnits, reviveDeadLetteredSyncUnits, upsertSyncUnits } from './syncUnits' +import { + claimDueUnits, + listShadowDiffUnits, + renewRunningUnitLease, + reviveDeadLetteredSyncUnits, + upsertSyncUnits, +} from './syncUnits' const test = withQx(base) @@ -291,3 +297,70 @@ describe('reviveDeadLetteredSyncUnits', () => { expect(await reviveDeadLetteredSyncUnits(qx, integrationId, 'mentionPoll')).toBe(0) }) }) + +async function setClaimState( + qx: QueryExecutor, + id: string, + state: { lockedAgo: string | null; nextRunAgo: string }, +): Promise { + await qx.result( + `UPDATE integration.sync_units + SET "lockedAt" = now() - $(lockedAgo)::interval, + "nextRunAt" = now() - $(nextRunAgo)::interval + WHERE id = $(id)`, + { id, ...state }, + ) +} + +async function readClaimState( + qx: QueryExecutor, + id: string, +): Promise<{ lockedAgoSeconds: number | null; nextRunAt: string }> { + return qx.selectOne( + `SELECT EXTRACT(EPOCH FROM (now() - "lockedAt"))::float8 AS "lockedAgoSeconds", "nextRunAt" + FROM integration.sync_units WHERE id = $(id)`, + { id }, + ) +} + +describe('renewRunningUnitLease', () => { + test('renews the claim and pushes the next probe out', async ({ qx }) => { + const integrationId = await createIntegration(qx) + const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' }) + await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '10 minutes' }) + const nextRunAt = new Date(Date.now() + 5 * 60_000) + + expect(await renewRunningUnitLease(qx, id, nextRunAt)).toBe(1) + + const state = await readClaimState(qx, id) + expect(state.lockedAgoSeconds).toBeLessThan(5) + expect(new Date(state.nextRunAt).getTime()).toBe(nextRunAt.getTime()) + }) + + test('leaves a unit alone once its run has recorded its end', async ({ qx }) => { + const integrationId = await createIntegration(qx) + const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' }) + await setClaimState(qx, id, { lockedAgo: null, nextRunAgo: '-5 minutes' }) + const before = await readClaimState(qx, id) + + expect(await renewRunningUnitLease(qx, id, new Date(Date.now() + 60_000))).toBe(0) + + expect(await readClaimState(qx, id)).toEqual(before) + }) +}) + +describe('claimDueUnits', () => { + test('skips a due unit whose running lease was renewed, and reclaims it once the lease lapses', async ({ + qx, + }) => { + const integrationId = await createIntegration(qx) + const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' }) + await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '1 minute' }) + await renewRunningUnitLease(qx, id, new Date(Date.now() - 1000)) + + expect((await claimDueUnits(qx, 100)).map((u) => u.id)).not.toContain(id) + + await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '1 minute' }) + expect((await claimDueUnits(qx, 100)).map((u) => u.id)).toContain(id) + }) +}) diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index 3cf5485b6b..eb12138908 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -93,6 +93,21 @@ export async function rescheduleUnit( ) } +// Keeps a still-running unit claimed so lockedAt stays set for the whole run. A NULL lockedAt +// means the run already recorded its end after this claim, so its own nextRunAt is kept. +export async function renewRunningUnitLease( + qx: QueryExecutor, + id: string, + nextRunAt: Date, +): Promise { + return qx.result( + `UPDATE integration.sync_units + SET "nextRunAt" = $(nextRunAt), "lockedAt" = now(), "updatedAt" = now() + WHERE id = $(id) AND "lockedAt" IS NOT NULL`, + { id, nextRunAt }, + ) +} + export async function recordRunSuccess( qx: QueryExecutor, id: string, From f4851b6b2fb5c71512b8e377df31f1c89ad84012 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:50:34 +0200 Subject: [PATCH 07/13] fix: record a page-capped octolens poll as incomplete while keeping the regular cadence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/activities/syncRunActivities.test.ts | 122 ++++++++++++++++++ .../src/activities/syncRunActivities.ts | 5 +- .../octolens/syncs/mentionPoll.test.ts | 2 +- .../connectors/octolens/syncs/mentionPoll.ts | 3 +- services/libs/connectors/src/types.ts | 2 + 5 files changed, 130 insertions(+), 4 deletions(-) create mode 100644 services/apps/connectors_worker/src/activities/syncRunActivities.test.ts diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts new file mode 100644 index 0000000000..a0290c5b4f --- /dev/null +++ b/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts @@ -0,0 +1,122 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import type { SyncContext, SyncOutcome } from '@crowd/connectors' + +import { executeSync } from './syncRunActivities' + +const CADENCE_RUN_AT = new Date('2026-10-16T12:05:00.000Z') +const SHORT_DEFER_RUN_AT = new Date('2026-10-16T12:01:00.000Z') + +const mocks = vi.hoisted(() => ({ + run: vi.fn(), + getUnitById: vi.fn(), + recordRunSuccess: vi.fn(), +})) + +vi.mock('@temporalio/activity', () => ({ + Context: { + current: () => ({ + info: { workflowExecution: { workflowId: 'wf', runId: 'run' } }, + heartbeat: vi.fn(), + }), + }, +})) + +vi.mock('../main', () => ({ + svc: { + postgres: { writer: {} }, + redis: {}, + dataSinkWorkerEmitter: {}, + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }, +})) + +vi.mock('@crowd/logging', async (importOriginal) => ({ + ...(await importOriginal()), + getChildLogger: () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }), +})) + +vi.mock('@crowd/connectors', async (importOriginal) => ({ + ...(await importOriginal()), + getManifest: () => ({}), + getSync: () => ({ cadenceMinutes: 5, schema: {}, run: mocks.run }), + findSync: () => ({ cadenceMinutes: 5 }), + createTokenPool: () => ({ acquire: vi.fn(), park: vi.fn(), invalidate: vi.fn() }), + createHttpClient: () => ({ requestCount: () => 0 }), + createEmit: () => ({ emit: vi.fn(), emittedCount: () => 3 }), +})) + +vi.mock('@crowd/data-access-layer/src/connectors', () => ({ + getUnitById: mocks.getUnitById, + recordRunSuccess: mocks.recordRunSuccess, + recordRunPartial: vi.fn(), + recordRunFailure: vi.fn(), + parkUnit: vi.fn(), + recordShadowRecords: vi.fn(), +})) + +vi.mock('@crowd/data-access-layer/src/integrations', () => ({ + fetchIntegrationById: async () => ({ segmentId: 'segment-1' }), +})) + +vi.mock( + '@crowd/data-access-layer/src/old/apps/integration_stream_worker/integrationStream.repo', + () => ({ + default: class { + publishExternalResult = vi.fn() + }, + }), +) + +vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({ + dbStoreQx: vi.fn(() => ({})), +})) + +vi.mock('../scheduling', () => ({ + cadenceRunAt: () => CADENCE_RUN_AT, + shortDeferRunAt: () => SHORT_DEFER_RUN_AT, + failureRunAt: () => SHORT_DEFER_RUN_AT, +})) + +const UNIT = { + id: 'unit-1', + integrationId: 'integration-1', + platform: 'octolens', + syncName: 'mentionPoll', + channelId: 'channel', + channelName: 'channel', + watermark: { cursor: 'before' }, + emitEnabled: true, + consecutiveFailures: 0, +} + +function runReturning(outcome: SyncOutcome) { + mocks.run.mockImplementation(async (ctx: SyncContext) => { + await ctx.commitWatermark({ cursor: 'after' }) + return outcome + }) +} + +beforeEach(() => { + vi.clearAllMocks() + mocks.getUnitById.mockResolvedValue(UNIT) +}) + +describe('executeSync outcome scheduling', () => { + it.each([ + [{ complete: true }, true, CADENCE_RUN_AT], + [{ complete: false }, false, SHORT_DEFER_RUN_AT], + [{ complete: false, resumeAtCadence: true }, false, CADENCE_RUN_AT], + ])('records %j as complete=%s, next run at %s', async (outcome, complete, nextRunAt) => { + runReturning(outcome) + + await executeSync(UNIT.id) + + expect(mocks.recordRunSuccess).toHaveBeenCalledWith( + {}, + UNIT.id, + { watermark: { cursor: 'after' }, emittedCount: 3, complete }, + nextRunAt, + ) + }) +}) diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.ts index ca1f265594..f573640522 100644 --- a/services/apps/connectors_worker/src/activities/syncRunActivities.ts +++ b/services/apps/connectors_worker/src/activities/syncRunActivities.ts @@ -156,7 +156,10 @@ export async function executeSync(unitId: string): Promise { } const outcome = await sync.run(ctx) - const nextRunAt = outcome.complete ? cadenceRunAt(sync.cadenceMinutes) : shortDeferRunAt() + const nextRunAt = + outcome.complete || outcome.resumeAtCadence + ? cadenceRunAt(sync.cadenceMinutes) + : shortDeferRunAt() await recordRunSuccess( qx, diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts index 6fcc2fceaa..35d4a02031 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts @@ -740,7 +740,7 @@ describe('runMentionPoll page cap', () => { } const { ctx, watermarks } = makeCtx({ watermark: { startDate: START, backfill: QUEUED } }) - expect(await runMentionPoll(ctx)).toEqual({ complete: true }) + expect(await runMentionPoll(ctx)).toEqual({ complete: false, resumeAtCadence: true }) expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(20) expect(mockedFetchMentionsPage).not.toHaveBeenCalledWith('test-token', { keywordIds: [7], diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts index de27799550..e6d0516cc0 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts @@ -178,12 +178,11 @@ export async function runMentionPoll(ctx: SyncContext): Promise= MAX_POLL_PAGES_PER_RUN) { - // complete: true schedules the next run at the normal cadence; false would retry within ~1 minute. ctx.log.info( { pagesFetched }, 'octolens mention poll paused at the page cap until next cadence', ) - return { complete: true } + return { complete: false, resumeAtCadence: true } } if (!polled) { ctx.log.info({ pagesFetched }, 'octolens mention poll deferred: run budget exhausted') diff --git a/services/libs/connectors/src/types.ts b/services/libs/connectors/src/types.ts index bcd7b05751..003f404319 100644 --- a/services/libs/connectors/src/types.ts +++ b/services/libs/connectors/src/types.ts @@ -51,6 +51,8 @@ export interface SyncContext { export interface SyncOutcome { complete: boolean + // With complete: false, waits for the sync's normal cadence instead of resuming within a minute. + resumeAtCadence?: boolean } export interface SyncDefinition { From 6b3e0ad91db85d769b29fde0bf0b5bbdd53a1bc6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 14:54:03 +0200 Subject: [PATCH 08/13] refactor: rely on the sync unit lock alone in the octolens backfill script MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/activities/dispatcherActivities.ts | 16 ++++++++++++- .../src/bin/backfill-octolens-mentions.ts | 23 ++----------------- .../src/octolensBackfill.test.ts | 3 --- .../src/syncRunWorkflowId.ts | 19 --------------- .../data-access-layer/src/octolens/index.ts | 8 ++----- .../src/octolens/mentionPollWatermark.test.ts | 4 ++-- 6 files changed, 21 insertions(+), 52 deletions(-) delete mode 100644 services/apps/connectors_worker/src/syncRunWorkflowId.ts diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts index 507d58b4d1..e9a036fdf8 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -11,13 +11,13 @@ import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal import { svc } from '../main' import { runningProbeRunAt, shortDeferRunAt } from '../scheduling' -import { syncRunWorkflowId } from '../syncRunWorkflowId' import type { IAdmissionResult, IDispatchCounts, StartRunResult } from '../types' const TASK_QUEUE = 'connectors' const HEARTBEAT_TTL_SECONDS = 300 const DEFAULT_RUN_ESTIMATE = 50 const BUDGET_PROBE_CONCURRENCY = 10 +const CHANNEL_LABEL_MAX_LENGTH = 80 export async function claimDue(limit: number): Promise { return claimDueUnits(dbStoreQx(svc.postgres.writer), limit) @@ -48,6 +48,20 @@ export async function deferUnit(unitId: string): Promise { await rescheduleUnit(dbStoreQx(svc.postgres.writer), unitId, shortDeferRunAt()) } +function channelLabel(channelName: string): string { + const label = channelName + .replace(/^[a-z][a-z0-9+.-]*:\/\/[^/]+\//i, '') + .replace(/^\/+|\/+$/g, '') + .replace(/\s+/g, '-') + .slice(0, CHANNEL_LABEL_MAX_LENGTH) + + return label || 'unknown-channel' +} + +export function syncRunWorkflowId(unit: IClaimedUnit): string { + return `sync-run/${unit.platform}/${unit.syncName}/${channelLabel(unit.channelName)}/${unit.id}` +} + export async function startRun(unit: IClaimedUnit): Promise { try { await svc.temporal.workflow.start('syncRun', { diff --git a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts index 094b49a6bd..89f8a68da4 100644 --- a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts +++ b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts @@ -5,14 +5,10 @@ import { queueOctolensKeywordBackfills, rewindOctolensMentionPollStartDate, } from '@crowd/data-access-layer/src/octolens' -import type { - IOctolensMentionPollUnit, - IOctolensMentionPollWriteResult, -} from '@crowd/data-access-layer/src/octolens' +import type { IOctolensMentionPollWriteResult } from '@crowd/data-access-layer/src/octolens' import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' import { getServiceLogger } from '@crowd/logging' -import { TEMPORAL_CONFIG, WorkflowNotFoundError, getTemporalClient } from '@crowd/temporal' import { OCTOLENS_BACKFILL_USAGE, @@ -20,7 +16,6 @@ import { planOctolensBackfill, } from '../octolensBackfill' import type { OctolensBackfillArgs, OctolensBackfillPlan } from '../octolensBackfill' -import { syncRunWorkflowId } from '../syncRunWorkflowId' const log = getServiceLogger() @@ -48,20 +43,6 @@ async function loadPlan(qx: QueryExecutor, args: OctolensBackfillArgs) { return planOctolensBackfill(args, { units, activeMappingCounts }) } -// The dispatcher clears lockedAt when it re-claims a unit whose run outlived the claim lease, -// so only Temporal knows a long run is still going and will overwrite the watermark at its end. -async function isRunInFlight(unit: IOctolensMentionPollUnit): Promise { - const temporal = await getTemporalClient(TEMPORAL_CONFIG()) - const workflowId = syncRunWorkflowId(unit) - try { - const description = await temporal.workflow.getHandle(workflowId).describe() - return description.status.name === 'RUNNING' - } catch (err) { - if (err instanceof WorkflowNotFoundError) return false - throw err - } -} - function describePlan(plan: OctolensBackfillPlan): Record { const base = { unitId: plan.unit.id, integrationId: plan.unit.integrationId } if (plan.mode === 'keywords') { @@ -103,7 +84,7 @@ setImmediate(async () => { process.exit(0) } - const busy = plan.unit.lockedAt !== null || (await isRunInFlight(plan.unit)) + const busy = plan.unit.lockedAt !== null if (args.dryRun) { log.info({ ...summary, busy }, 'dry run: would update the octolens mention poll watermark') diff --git a/services/apps/connectors_worker/src/octolensBackfill.test.ts b/services/apps/connectors_worker/src/octolensBackfill.test.ts index a0ad57bade..35bf82170b 100644 --- a/services/apps/connectors_worker/src/octolensBackfill.test.ts +++ b/services/apps/connectors_worker/src/octolensBackfill.test.ts @@ -15,9 +15,6 @@ function unit(overrides: Partial = {}): IOctolensMenti return { id: 'unit-1', integrationId: 'anchor-1', - platform: 'octolens', - syncName: 'mentionPoll', - channelName: 'octolens-mentions', status: 'active', lockedAt: null, watermark: { diff --git a/services/apps/connectors_worker/src/syncRunWorkflowId.ts b/services/apps/connectors_worker/src/syncRunWorkflowId.ts deleted file mode 100644 index 471587ecd8..0000000000 --- a/services/apps/connectors_worker/src/syncRunWorkflowId.ts +++ /dev/null @@ -1,19 +0,0 @@ -import type { IClaimedUnit } from '@crowd/data-access-layer/src/connectors' - -const CHANNEL_LABEL_MAX_LENGTH = 80 - -function channelLabel(channelName: string): string { - const label = channelName - .replace(/^[a-z][a-z0-9+.-]*:\/\/[^/]+\//i, '') - .replace(/^\/+|\/+$/g, '') - .replace(/\s+/g, '-') - .slice(0, CHANNEL_LABEL_MAX_LENGTH) - - return label || 'unknown-channel' -} - -export function syncRunWorkflowId( - unit: Pick, -): string { - return `sync-run/${unit.platform}/${unit.syncName}/${channelLabel(unit.channelName)}/${unit.id}` -} diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts index ff4448cdde..c5e3ce8ee6 100644 --- a/services/libs/data-access-layer/src/octolens/index.ts +++ b/services/libs/data-access-layer/src/octolens/index.ts @@ -279,9 +279,6 @@ const MENTION_POLL_SYNC_NAME = 'mentionPoll' export interface IOctolensMentionPollUnit { id: string integrationId: string - platform: string - syncName: string - channelName: string status: string lockedAt: Date | null watermark: Record | null @@ -299,8 +296,7 @@ export async function listOctolensMentionPollUnits( ): Promise { return qx.select( ` - SELECT su.id, su."integrationId", su.platform, su."syncName", su."channelName", su.status, - su."lockedAt", su.watermark + SELECT su.id, su."integrationId", su.status, su."lockedAt", su.watermark FROM integration.sync_units su JOIN public.integrations i ON i.id = su."integrationId" WHERE su.platform = $(platform) @@ -319,7 +315,7 @@ const STORED_WATERMARK = `(CASE WHEN jsonb_typeof(su.watermark) = 'object' THEN const STORED_BACKFILL = `(CASE WHEN jsonb_typeof(su.watermark->'backfill') = 'array' THEN su.watermark->'backfill' ELSE '[]'::jsonb END)` // A run keeps the watermark in memory and overwrites it when it ends, so this only writes while -// the unit is unclaimed; callers must separately rule out a run still in flight with no claim. +// the unit is unclaimed; the dispatcher keeps lockedAt set until the run records its end. async function updateIdleMentionPollWatermark( qx: QueryExecutor, unitId: string, diff --git a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts index 782a808fc2..482e0b9d2a 100644 --- a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts +++ b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts @@ -81,8 +81,8 @@ describe('listOctolensMentionPollUnits', () => { await createMentionPollUnit(qx, deletedAnchorId) const units = await listOctolensMentionPollUnits(qx) - expect(units.map((u) => [u.id, u.integrationId, u.platform, u.syncName, u.status])).toEqual([ - [unitId, anchorId, 'octolens', 'mentionPoll', 'dead_letter'], + expect(units.map((u) => [u.id, u.integrationId, u.status])).toEqual([ + [unitId, anchorId, 'dead_letter'], ]) }) }) From e81ac8cbdd06893ac07d992b10926ef38c69bb3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 15:11:12 +0200 Subject: [PATCH 09/13] fix: keep the claim when deferring a unit whose previous lease lapsed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../activities/dispatcherActivities.test.ts | 19 ++++++++++++++++++- .../src/activities/dispatcherActivities.ts | 10 ++++++++-- .../src/workflows/dispatcher.ts | 2 +- .../src/connectors/syncUnits.test.ts | 17 +++++++++++++++++ .../src/connectors/syncUnits.ts | 10 ++++++---- .../data-access-layer/src/connectors/types.ts | 2 +- 6 files changed, 51 insertions(+), 9 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts index e1cb8613d5..051b52d6ef 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts @@ -1,6 +1,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' -import { startRun } from './dispatcherActivities' +import { deferUnit, startRun } from './dispatcherActivities' const PROBE_AT = new Date('2026-10-16T12:05:00.000Z') @@ -43,6 +43,7 @@ const UNIT = { syncName: 'issues', channelId: 'repo', channelName: 'https://github.com/org/repo', + leaseLapsed: false, } beforeEach(() => { @@ -70,3 +71,19 @@ describe('startRun', () => { expect(mocks.rescheduleUnit).not.toHaveBeenCalled() }) }) + +describe('deferUnit', () => { + it('releases a freshly claimed unit', async () => { + await deferUnit(UNIT.id, false) + + expect(mocks.rescheduleUnit).toHaveBeenCalledWith({}, UNIT.id, PROBE_AT) + expect(mocks.renewRunningUnitLease).not.toHaveBeenCalled() + }) + + it('keeps the claim on a unit whose previous lease lapsed, as its run may still be going', async () => { + await deferUnit(UNIT.id, true) + + expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id, PROBE_AT) + expect(mocks.rescheduleUnit).not.toHaveBeenCalled() + }) +}) diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts index e9a036fdf8..f3d68647d7 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -44,8 +44,14 @@ export async function admitByBudget(units: IClaimedUnit[]): Promise { - await rescheduleUnit(dbStoreQx(svc.postgres.writer), unitId, shortDeferRunAt()) +// A lapsed lease may belong to a run still in flight, so that claim is kept instead of released. +export async function deferUnit(unitId: string, leaseLapsed = false): Promise { + const qx = dbStoreQx(svc.postgres.writer) + if (leaseLapsed) { + await renewRunningUnitLease(qx, unitId, shortDeferRunAt()) + } else { + await rescheduleUnit(qx, unitId, shortDeferRunAt()) + } } function channelLabel(channelName: string): string { diff --git a/services/apps/connectors_worker/src/workflows/dispatcher.ts b/services/apps/connectors_worker/src/workflows/dispatcher.ts index 7cd25d05f2..c13ce7a96c 100644 --- a/services/apps/connectors_worker/src/workflows/dispatcher.ts +++ b/services/apps/connectors_worker/src/workflows/dispatcher.ts @@ -36,7 +36,7 @@ export async function dispatcher(): Promise { } for (const unit of deferred) { - await activity.deferUnit(unit.id) + await activity.deferUnit(unit.id, unit.leaseLapsed) } await activity.logDispatchSummary({ 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 a69080b6d6..324f351402 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -350,6 +350,23 @@ describe('renewRunningUnitLease', () => { }) describe('claimDueUnits', () => { + test('reports whether the claim took over a lapsed lease', async ({ qx }) => { + const integrationId = await createIntegration(qx) + const freeId = await createSyncUnit(qx, integrationId, { channelName: 'a', syncName: 'issues' }) + const lapsedId = await createSyncUnit(qx, integrationId, { + channelName: 'b', + syncName: 'issues', + }) + await setClaimState(qx, freeId, { lockedAgo: null, nextRunAgo: '1 minute' }) + await setClaimState(qx, lapsedId, { lockedAgo: '6 minutes', nextRunAgo: '1 minute' }) + + const claimed = await claimDueUnits(qx, 100) + + expect(claimed.find((u) => u.id === freeId)?.leaseLapsed).toBe(false) + expect(claimed.find((u) => u.id === lapsedId)?.leaseLapsed).toBe(true) + expect((await readClaimState(qx, lapsedId)).lockedAgoSeconds).toBeLessThan(5) + }) + test('skips a due unit whose running lease was renewed, and reclaims it once the lease lapses', async ({ qx, }) => { diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index eb12138908..6a808a72d2 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -60,8 +60,8 @@ export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise +> & { leaseLapsed: boolean } export type IShadowDiffUnit = Pick From eb92040c3dd2cdb241f52ea1135a5a7a3c2c426f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 15:18:17 +0200 Subject: [PATCH 10/13] fix: re-lock a sync unit when its run starts so no run executes unlocked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/activities/syncRunActivities.test.ts | 15 +++++++++++++++ .../src/activities/syncRunActivities.ts | 2 ++ .../src/connectors/syncUnits.test.ts | 14 ++++++++++++++ .../data-access-layer/src/connectors/syncUnits.ts | 11 +++++++++++ 4 files changed, 42 insertions(+) diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts index a0290c5b4f..ea6a531fe7 100644 --- a/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts @@ -10,6 +10,7 @@ const SHORT_DEFER_RUN_AT = new Date('2026-10-16T12:01:00.000Z') const mocks = vi.hoisted(() => ({ run: vi.fn(), getUnitById: vi.fn(), + reassertUnitClaim: vi.fn(), recordRunSuccess: vi.fn(), })) @@ -48,6 +49,7 @@ vi.mock('@crowd/connectors', async (importOriginal) => ({ vi.mock('@crowd/data-access-layer/src/connectors', () => ({ getUnitById: mocks.getUnitById, + reassertUnitClaim: mocks.reassertUnitClaim, recordRunSuccess: mocks.recordRunSuccess, recordRunPartial: vi.fn(), recordRunFailure: vi.fn(), @@ -120,3 +122,16 @@ describe('executeSync outcome scheduling', () => { ) }) }) + +describe('executeSync claim', () => { + it('re-locks the unit before reading it, so the whole run is covered by lockedAt', async () => { + runReturning({ complete: true }) + + await executeSync(UNIT.id) + + expect(mocks.reassertUnitClaim).toHaveBeenCalledWith({}, UNIT.id) + expect(mocks.reassertUnitClaim.mock.invocationCallOrder[0]).toBeLessThan( + mocks.getUnitById.mock.invocationCallOrder[0], + ) + }) +}) diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.ts index f573640522..1fa0d37c37 100644 --- a/services/apps/connectors_worker/src/activities/syncRunActivities.ts +++ b/services/apps/connectors_worker/src/activities/syncRunActivities.ts @@ -14,6 +14,7 @@ import type { ConnectorHttp, Emitter, SyncContext } from '@crowd/connectors' import { getUnitById, parkUnit, + reassertUnitClaim, recordRunFailure, recordRunPartial, recordRunSuccess, @@ -39,6 +40,7 @@ const PARKED_ERROR_CLASSES = ['provider.rate_limit', 'provider.unavailable'] as export async function executeSync(unitId: string): Promise { const qx = dbStoreQx(svc.postgres.writer) + await reassertUnitClaim(qx, unitId) const unit = await getUnitById(qx, unitId) if (!unit) { throw new Error(`sync unit ${unitId} not found`) 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 324f351402..1924a4280b 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -7,6 +7,7 @@ import { withQx } from '@crowd/test-kit/db' import { claimDueUnits, listShadowDiffUnits, + reassertUnitClaim, renewRunningUnitLease, reviveDeadLetteredSyncUnits, upsertSyncUnits, @@ -381,3 +382,16 @@ describe('claimDueUnits', () => { expect((await claimDueUnits(qx, 100)).map((u) => u.id)).toContain(id) }) }) + +describe('reassertUnitClaim', () => { + test('locks a unit whose lock a previous run cleared', async ({ qx }) => { + const integrationId = await createIntegration(qx) + const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' }) + await setClaimState(qx, id, { lockedAgo: null, nextRunAgo: '1 minute' }) + + await reassertUnitClaim(qx, id) + + expect((await readClaimState(qx, id)).lockedAgoSeconds).toBeLessThan(5) + expect((await claimDueUnits(qx, 100)).map((u) => u.id)).not.toContain(id) + }) +}) diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index 6a808a72d2..1eb6d73daf 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -110,6 +110,17 @@ export async function renewRunningUnitLease( ) } +// A previous run's end write can clear the lock between this run's claim and its start, so +// every run re-locks before reading its unit and stays locked until its own end write. +export async function reassertUnitClaim(qx: QueryExecutor, id: string): Promise { + await qx.result( + `UPDATE integration.sync_units + SET "lockedAt" = now(), "updatedAt" = now() + WHERE id = $(id)`, + { id }, + ) +} + export async function recordRunSuccess( qx: QueryExecutor, id: string, From 324b2cb31c802f55a6e7b6d14bfbd2b3bfc58dae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 15:29:59 +0200 Subject: [PATCH 11/13] refactor: point renewed sync unit leases at the lease end and state the resume flag as an invariant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../activities/dispatcherActivities.test.ts | 15 ++++++------- .../src/activities/dispatcherActivities.ts | 20 +++++------------- .../apps/connectors_worker/src/scheduling.ts | 7 ------- services/libs/connectors/src/types.ts | 2 +- .../src/connectors/syncUnits.test.ts | 21 ++++++++++++------- .../src/connectors/syncUnits.ts | 12 +++++------ 6 files changed, 30 insertions(+), 47 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts index 051b52d6ef..cae9bba76b 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts @@ -2,11 +2,10 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { deferUnit, startRun } from './dispatcherActivities' -const PROBE_AT = new Date('2026-10-16T12:05:00.000Z') +const SHORT_DEFER_AT = new Date('2026-10-16T12:01:00.000Z') const mocks = vi.hoisted(() => ({ start: vi.fn(), - describe: vi.fn(), renewRunningUnitLease: vi.fn(), rescheduleUnit: vi.fn(), })) @@ -16,7 +15,7 @@ vi.mock('../main', () => ({ postgres: { writer: {} }, log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, temporal: { - workflow: { start: mocks.start, getHandle: () => ({ describe: mocks.describe }) }, + workflow: { start: mocks.start }, }, }, })) @@ -32,8 +31,7 @@ vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({ })) vi.mock('../scheduling', () => ({ - runningProbeRunAt: () => PROBE_AT, - shortDeferRunAt: () => PROBE_AT, + shortDeferRunAt: () => SHORT_DEFER_AT, })) const UNIT = { @@ -48,7 +46,6 @@ const UNIT = { beforeEach(() => { vi.clearAllMocks() - mocks.describe.mockResolvedValue({ startTime: new Date('2026-10-16T11:00:00.000Z') }) mocks.renewRunningUnitLease.mockResolvedValue(1) }) @@ -67,7 +64,7 @@ describe('startRun', () => { mocks.start.mockRejectedValue(alreadyStarted) expect(await startRun(UNIT)).toBe('alreadyRunning') - expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id, PROBE_AT) + expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id) expect(mocks.rescheduleUnit).not.toHaveBeenCalled() }) }) @@ -76,14 +73,14 @@ describe('deferUnit', () => { it('releases a freshly claimed unit', async () => { await deferUnit(UNIT.id, false) - expect(mocks.rescheduleUnit).toHaveBeenCalledWith({}, UNIT.id, PROBE_AT) + expect(mocks.rescheduleUnit).toHaveBeenCalledWith({}, UNIT.id, SHORT_DEFER_AT) expect(mocks.renewRunningUnitLease).not.toHaveBeenCalled() }) it('keeps the claim on a unit whose previous lease lapsed, as its run may still be going', async () => { await deferUnit(UNIT.id, true) - expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id, PROBE_AT) + expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id) expect(mocks.rescheduleUnit).not.toHaveBeenCalled() }) }) diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts index f3d68647d7..0a467a6d43 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -10,7 +10,7 @@ import { RedisCache } from '@crowd/redis' import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal' import { svc } from '../main' -import { runningProbeRunAt, shortDeferRunAt } from '../scheduling' +import { shortDeferRunAt } from '../scheduling' import type { IAdmissionResult, IDispatchCounts, StartRunResult } from '../types' const TASK_QUEUE = 'connectors' @@ -48,7 +48,7 @@ export async function admitByBudget(units: IClaimedUnit[]): Promise { const qx = dbStoreQx(svc.postgres.writer) if (leaseLapsed) { - await renewRunningUnitLease(qx, unitId, shortDeferRunAt()) + await renewRunningUnitLease(qx, unitId) } else { await rescheduleUnit(qx, unitId, shortDeferRunAt()) } @@ -96,23 +96,13 @@ export async function startRun(unit: IClaimedUnit): Promise { } async function backOffRunningUnit(unit: IClaimedUnit): Promise { - const nextRunAt = await runningSince(unit) - .then(runningProbeRunAt) - .catch(() => shortDeferRunAt()) - - const renewed = await renewRunningUnitLease(dbStoreQx(svc.postgres.writer), unit.id, nextRunAt) + const renewed = await renewRunningUnitLease(dbStoreQx(svc.postgres.writer), unit.id) svc.log.info( - { unitId: unit.id, nextRunAt, renewed: renewed > 0 }, - 'unit already running, lease renewed until the next probe', + { unitId: unit.id, renewed: renewed > 0 }, + 'unit already running, lease renewed until it lapses', ) } -async function runningSince(unit: IClaimedUnit): Promise { - const description = await svc.temporal.workflow.getHandle(syncRunWorkflowId(unit)).describe() - - return description.startTime -} - export async function logDispatchSummary(counts: IDispatchCounts): Promise { svc.log.info({ event: 'dispatcher_tick', ...counts }, 'dispatch summary') } diff --git a/services/apps/connectors_worker/src/scheduling.ts b/services/apps/connectors_worker/src/scheduling.ts index ee5c886e07..6d742d16ae 100644 --- a/services/apps/connectors_worker/src/scheduling.ts +++ b/services/apps/connectors_worker/src/scheduling.ts @@ -2,7 +2,6 @@ const JITTER_RATIO = 0.1 const SHORT_DEFER_MIN_MS = 30_000 const SHORT_DEFER_JITTER_MS = 60_000 const FAILURE_BACKOFF_BASE_MS = 60_000 -const RUNNING_PROBE_CAP_MS = 5 * 60_000 const FALLBACK_FAILURE_CAP_MS = 60 * 60_000 function withJitter(delayMs: number): number { @@ -17,12 +16,6 @@ export function shortDeferRunAt(): Date { return new Date(Date.now() + SHORT_DEFER_MIN_MS + Math.random() * SHORT_DEFER_JITTER_MS) } -export function runningProbeRunAt(startedAt: Date): Date { - const elapsedMs = Math.max(SHORT_DEFER_MIN_MS, Date.now() - startedAt.getTime()) - - return new Date(Date.now() + withJitter(Math.min(elapsedMs, RUNNING_PROBE_CAP_MS))) -} - export function failureRunAt(consecutiveFailures: number, cadenceMinutes: number | null): Date { const capMs = cadenceMinutes === null ? FALLBACK_FAILURE_CAP_MS : cadenceMinutes * 60_000 const backoffMs = FAILURE_BACKOFF_BASE_MS * 2 ** (Math.max(1, consecutiveFailures) - 1) diff --git a/services/libs/connectors/src/types.ts b/services/libs/connectors/src/types.ts index 003f404319..2803342bc5 100644 --- a/services/libs/connectors/src/types.ts +++ b/services/libs/connectors/src/types.ts @@ -51,7 +51,7 @@ export interface SyncContext { export interface SyncOutcome { complete: boolean - // With complete: false, waits for the sync's normal cadence instead of resuming within a minute. + // Only meaningful with complete: false; the next run then waits for the sync's normal cadence. resumeAtCadence?: boolean } 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 1924a4280b..e8140c33c3 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -325,17 +325,21 @@ async function readClaimState( } describe('renewRunningUnitLease', () => { - test('renews the claim and pushes the next probe out', async ({ qx }) => { + test('renews the claim and moves the next run to the end of the new lease', async ({ qx }) => { const integrationId = await createIntegration(qx) const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' }) await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '10 minutes' }) - const nextRunAt = new Date(Date.now() + 5 * 60_000) - expect(await renewRunningUnitLease(qx, id, nextRunAt)).toBe(1) + expect(await renewRunningUnitLease(qx, id)).toBe(1) - const state = await readClaimState(qx, id) - expect(state.lockedAgoSeconds).toBeLessThan(5) - expect(new Date(state.nextRunAt).getTime()).toBe(nextRunAt.getTime()) + const lease: { lockedAgoSeconds: number; leaseSeconds: number } = await qx.selectOne( + `SELECT EXTRACT(EPOCH FROM (now() - "lockedAt"))::float8 AS "lockedAgoSeconds", + EXTRACT(EPOCH FROM ("nextRunAt" - "lockedAt"))::float8 AS "leaseSeconds" + FROM integration.sync_units WHERE id = $(id)`, + { id }, + ) + expect(lease.lockedAgoSeconds).toBeLessThan(5) + expect(lease.leaseSeconds).toBe(300) }) test('leaves a unit alone once its run has recorded its end', async ({ qx }) => { @@ -344,7 +348,7 @@ describe('renewRunningUnitLease', () => { await setClaimState(qx, id, { lockedAgo: null, nextRunAgo: '-5 minutes' }) const before = await readClaimState(qx, id) - expect(await renewRunningUnitLease(qx, id, new Date(Date.now() + 60_000))).toBe(0) + expect(await renewRunningUnitLease(qx, id)).toBe(0) expect(await readClaimState(qx, id)).toEqual(before) }) @@ -374,7 +378,8 @@ describe('claimDueUnits', () => { const integrationId = await createIntegration(qx) const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' }) await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '1 minute' }) - await renewRunningUnitLease(qx, id, new Date(Date.now() - 1000)) + await renewRunningUnitLease(qx, id) + await setClaimState(qx, id, { lockedAgo: '0 minutes', nextRunAgo: '1 minute' }) expect((await claimDueUnits(qx, 100)).map((u) => u.id)).not.toContain(id) diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index 1eb6d73daf..207508d81f 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -97,16 +97,14 @@ export async function rescheduleUnit( // Keeps a still-running unit claimed so lockedAt stays set for the whole run. A NULL lockedAt // means the run already recorded its end after this claim, so its own nextRunAt is kept. -export async function renewRunningUnitLease( - qx: QueryExecutor, - id: string, - nextRunAt: Date, -): Promise { +export async function renewRunningUnitLease(qx: QueryExecutor, id: string): Promise { return qx.result( `UPDATE integration.sync_units - SET "nextRunAt" = $(nextRunAt), "lockedAt" = now(), "updatedAt" = now() + SET "lockedAt" = now(), + "nextRunAt" = now() + $(leaseMinutes) * interval '1 minute', + "updatedAt" = now() WHERE id = $(id) AND "lockedAt" IS NOT NULL`, - { id, nextRunAt }, + { id, leaseMinutes: CLAIM_LEASE_MINUTES }, ) } From 14cc77d44d40128a7d810e08bae9e04c779ca58b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Sat, 17 Oct 2026 20:21:03 +0200 Subject: [PATCH 12/13] fix: list each repeated --keywords id once in the octolens backfill plan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- services/apps/connectors_worker/src/octolensBackfill.test.ts | 4 ++-- services/apps/connectors_worker/src/octolensBackfill.ts | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/services/apps/connectors_worker/src/octolensBackfill.test.ts b/services/apps/connectors_worker/src/octolensBackfill.test.ts index 35bf82170b..bdb6505976 100644 --- a/services/apps/connectors_worker/src/octolensBackfill.test.ts +++ b/services/apps/connectors_worker/src/octolensBackfill.test.ts @@ -40,10 +40,10 @@ function state(overrides: Partial = {}): IOctolensBackfi } describe('parseOctolensBackfillArgs', () => { - it('parses keyword ids, dropping the pnpm separator', () => { + it('parses keyword ids, dropping the pnpm separator and repeated ids', () => { expect(parse('--', '--keywords', '9,8, 9', '--dry-run')).toEqual({ mode: 'keywords', - keywordIds: [9, 8, 9], + keywordIds: [9, 8], dryRun: true, }) }) diff --git a/services/apps/connectors_worker/src/octolensBackfill.ts b/services/apps/connectors_worker/src/octolensBackfill.ts index b926dbd349..a1de4b9d2b 100644 --- a/services/apps/connectors_worker/src/octolensBackfill.ts +++ b/services/apps/connectors_worker/src/octolensBackfill.ts @@ -68,8 +68,9 @@ export function parseOctolensBackfillArgs(argv: string[], now: Date): OctolensBa return { mode: 'from', startDate: parseStartDate(from as string, now), dryRun } } +// Repeats collapse to one entry so the plan and its log list each keyword once. function parseKeywordIds(raw: string): number[] { - return raw.split(',').map((part) => { + const ids = raw.split(',').map((part) => { const trimmed = part.trim() const id = Number(trimmed) if (!/^\d+$/.test(trimmed) || id < 1 || id > MAX_POSTGRES_INT) { @@ -77,6 +78,7 @@ function parseKeywordIds(raw: string): number[] { } return id }) + return [...new Set(ids)] } function parseStartDate(raw: string, now: Date): string { From cef0431170e00c8546e5b83e1a835c77da1911ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Sun, 18 Oct 2026 10:04:05 +0200 Subject: [PATCH 13/13] fix: ignore unusable backfill entries and refuse rewinds planned on a stale start date MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/bin/backfill-octolens-mentions.ts | 14 +++- .../src/octolensBackfill.test.ts | 13 ++++ .../connectors_worker/src/octolensBackfill.ts | 7 +- .../data-access-layer/src/octolens/index.ts | 38 +++++++--- .../src/octolens/mentionPollWatermark.test.ts | 72 ++++++++++++++++++- 5 files changed, 130 insertions(+), 14 deletions(-) diff --git a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts index 89f8a68da4..32b3300ddd 100644 --- a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts +++ b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts @@ -69,7 +69,12 @@ function applyPlan( if (plan.mode === 'keywords') { return queueOctolensKeywordBackfills(qx, plan.unit.id, plan.keywordIdsToQueue) } - return rewindOctolensMentionPollStartDate(qx, plan.unit.id, plan.startDate) + return rewindOctolensMentionPollStartDate( + qx, + plan.unit.id, + plan.startDate, + plan.previousStartDate, + ) } setImmediate(async () => { @@ -104,6 +109,13 @@ setImmediate(async () => { log.error(summary, `refused: a mention poll run claimed the unit just now; ${RETRY_HINT}`) process.exit(2) } + if (result.outcome === 'changed') { + log.error( + { ...summary, watermark: result.watermark }, + 'refused: the start date changed since the plan was checked (another rewind or the first poll run); re-run to check against the new start date', + ) + process.exit(2) + } if (result.outcome !== 'applied') { log.error( { ...summary, outcome: result.outcome }, diff --git a/services/apps/connectors_worker/src/octolensBackfill.test.ts b/services/apps/connectors_worker/src/octolensBackfill.test.ts index bdb6505976..494a704155 100644 --- a/services/apps/connectors_worker/src/octolensBackfill.test.ts +++ b/services/apps/connectors_worker/src/octolensBackfill.test.ts @@ -93,6 +93,19 @@ describe('planOctolensBackfill', () => { expect(plan).toMatchObject({ mode: 'keywords', keywordIdsToQueue: [9], alreadyQueued: [7] }) }) + it('queues keywords whose only queue entry is one the poll would drop', () => { + const watermark = { + startDate: '2026-10-16T11:55:00.000Z', + backfill: [{ keywordId: 7 }, { keywordId: 8, cursor: 42 }, { keywordId: 9, cursor: null }], + } + const plan = planOctolensBackfill( + parse('--keywords', '7,8,9'), + state({ units: [unit({ watermark })] }), + ) + + expect(plan).toMatchObject({ keywordIdsToQueue: [7, 8], alreadyQueued: [9] }) + }) + it('accepts --from exactly one year back', () => { expect(parse('--from', '2025-10-16T12:00:00Z')).toMatchObject({ startDate: '2025-10-16T12:00:00.000Z', diff --git a/services/apps/connectors_worker/src/octolensBackfill.ts b/services/apps/connectors_worker/src/octolensBackfill.ts index a1de4b9d2b..58cb9a1447 100644 --- a/services/apps/connectors_worker/src/octolensBackfill.ts +++ b/services/apps/connectors_worker/src/octolensBackfill.ts @@ -160,7 +160,12 @@ export function planOctolensBackfill( function queuedKeywordIds(watermark: Record | null): number[] { const backfill = watermark?.backfill if (!Array.isArray(backfill)) return [] + // Matches the poll, which drops entries whose cursor is missing or not a string. return backfill - .map((item) => (item as { keywordId?: unknown } | null)?.keywordId) + .filter((item) => { + const cursor = (item as { cursor?: unknown } | null)?.cursor + return cursor === null || typeof cursor === 'string' + }) + .map((item) => (item as { keywordId?: unknown }).keywordId) .filter((id): id is number => Number.isInteger(id)) } diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts index c5e3ce8ee6..0ca1652dcc 100644 --- a/services/libs/data-access-layer/src/octolens/index.ts +++ b/services/libs/data-access-layer/src/octolens/index.ts @@ -284,7 +284,13 @@ export interface IOctolensMentionPollUnit { watermark: Record | null } -export type OctolensMentionPollWriteOutcome = 'applied' | 'locked' | 'inactive' | 'missing' +// 'changed': the unit was idle but its watermark no longer matched what the write was planned on. +export type OctolensMentionPollWriteOutcome = + | 'applied' + | 'locked' + | 'changed' + | 'inactive' + | 'missing' export interface IOctolensMentionPollWriteResult { outcome: OctolensMentionPollWriteOutcome @@ -321,6 +327,7 @@ async function updateIdleMentionPollWatermark( unitId: string, watermarkSql: string, params: Record, + guardSql = 'true', ): Promise { const filter = { unitId, platform: PlatformType.OCTOLENS, syncName: MENTION_POLL_SYNC_NAME } const updated: { watermark: Record }[] = await qx.select( @@ -332,6 +339,7 @@ async function updateIdleMentionPollWatermark( AND su."syncName" = $(syncName) AND su.status = 'active' AND su."lockedAt" IS NULL + AND ${guardSql} RETURNING su.watermark `, { ...params, ...filter }, @@ -340,19 +348,25 @@ async function updateIdleMentionPollWatermark( return { outcome: 'applied', watermark: updated[0].watermark } } - const unit: { status: string; watermark: Record | null } | null = - await qx.selectOneOrNone( - ` - SELECT su.status, su.watermark + const unit: { + status: string + lockedAt: Date | null + watermark: Record | null + } | null = await qx.selectOneOrNone( + ` + SELECT su.status, su."lockedAt", su.watermark FROM integration.sync_units su WHERE su.id = $(unitId) AND su.platform = $(platform) AND su."syncName" = $(syncName) `, - filter, - ) + filter, + ) if (!unit) { return { outcome: 'missing', watermark: null } } - return { outcome: unit.status === 'active' ? 'locked' : 'inactive', watermark: unit.watermark } + if (unit.status !== 'active') { + return { outcome: 'inactive', watermark: unit.watermark } + } + return { outcome: unit.lockedAt ? 'locked' : 'changed', watermark: unit.watermark } } export async function queueOctolensKeywordBackfills( @@ -373,6 +387,8 @@ export async function queueOctolensKeywordBackfills( WHERE NOT EXISTS ( SELECT 1 FROM jsonb_array_elements(${STORED_BACKFILL}) e WHERE e->'keywordId' = to_jsonb(k.id) + -- The poll drops entries whose cursor is missing or not a string, so they don't count. + AND jsonb_typeof(e->'cursor') IN ('null', 'string') ) ), '[]'::jsonb @@ -386,6 +402,7 @@ export async function rewindOctolensMentionPollStartDate( qx: QueryExecutor, unitId: string, startDate: string, + expectedStartDate: string | null, ): Promise { return updateIdleMentionPollWatermark( qx, @@ -395,6 +412,9 @@ export async function rewindOctolensMentionPollStartDate( 'cursor', NULL, 'pendingMaxTimestamp', NULL )`, - { startDate }, + { startDate, expectedStartDate }, + // Only over the start date the rewind was validated against, so a concurrent rewind or the + // first poll run can't be silently overwritten. + `(${STORED_WATERMARK}->>'startDate') IS NOT DISTINCT FROM $(expectedStartDate)`, ) } diff --git a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts index 482e0b9d2a..066a1f79d8 100644 --- a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts +++ b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts @@ -108,6 +108,20 @@ describe('queueOctolensKeywordBackfills', () => { }) }) + test('queues a keyword whose only entry is one the poll would drop', async ({ qx }) => { + const malformed = [{ keywordId: 7 }, { keywordId: 8, cursor: 42 }] + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + watermark: { ...MID_CYCLE, backfill: malformed }, + }) + + const result = await queueOctolensKeywordBackfills(qx, unitId, [7, 8]) + + expect(result.outcome).toBe('applied') + expect(await readWatermark(qx, unitId)).toMatchObject({ + backfill: [...malformed, { keywordId: 7, cursor: null }, { keywordId: 8, cursor: null }], + }) + }) + test('creates the queue on a unit that has never run', async ({ qx }) => { const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx)) @@ -186,7 +200,12 @@ describe('rewindOctolensMentionPollStartDate', () => { watermark: MID_CYCLE, }) - const result = await rewindOctolensMentionPollStartDate(qx, unitId, '2026-09-01T00:00:00.000Z') + const result = await rewindOctolensMentionPollStartDate( + qx, + unitId, + '2026-09-01T00:00:00.000Z', + MID_CYCLE.startDate, + ) expect(result.outcome).toBe('applied') expect(await readWatermark(qx, unitId)).toEqual({ @@ -197,13 +216,55 @@ describe('rewindOctolensMentionPollStartDate', () => { }) }) + test('refuses when the start date changed since the rewind was planned', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { + watermark: MID_CYCLE, + }) + const first = await rewindOctolensMentionPollStartDate( + qx, + unitId, + '2026-08-01T00:00:00.000Z', + MID_CYCLE.startDate, + ) + + const second = await rewindOctolensMentionPollStartDate( + qx, + unitId, + '2026-09-01T00:00:00.000Z', + MID_CYCLE.startDate, + ) + + expect(first.outcome).toBe('applied') + expect(second.outcome).toBe('changed') + expect(await readWatermark(qx, unitId)).toMatchObject({ startDate: '2026-08-01T00:00:00.000Z' }) + }) + + test('rewinds a unit that has never run when it still has no start date', async ({ qx }) => { + const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx)) + + const result = await rewindOctolensMentionPollStartDate( + qx, + unitId, + '2026-09-01T00:00:00.000Z', + null, + ) + + expect(result.outcome).toBe('applied') + expect(await readWatermark(qx, unitId)).toMatchObject({ startDate: '2026-09-01T00:00:00.000Z' }) + }) + test('refuses and writes nothing while the unit is claimed', async ({ qx }) => { const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), { locked: true, watermark: MID_CYCLE, }) - const result = await rewindOctolensMentionPollStartDate(qx, unitId, '2026-09-01T00:00:00.000Z') + const result = await rewindOctolensMentionPollStartDate( + qx, + unitId, + '2026-09-01T00:00:00.000Z', + MID_CYCLE.startDate, + ) expect(result.outcome).toBe('locked') expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE) @@ -215,7 +276,12 @@ describe('rewindOctolensMentionPollStartDate', () => { watermark: MID_CYCLE, }) - const result = await rewindOctolensMentionPollStartDate(qx, unitId, '2026-09-01T00:00:00.000Z') + const result = await rewindOctolensMentionPollStartDate( + qx, + unitId, + '2026-09-01T00:00:00.000Z', + MID_CYCLE.startDate, + ) expect(result.outcome).toBe('inactive') expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE)