feat: add a guarded octolens backfill script and keep running sync units locked

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:*",

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

@@ -0,0 +1,86 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { deferUnit, startRun } from './dispatcherActivities'
+
+const SHORT_DEFER_AT = new Date('2026-10-16T12:01:00.000Z')
+
+const mocks = vi.hoisted(() => ({
+  start: 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 },
+    },
+  },
+}))
+
+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', () => ({
+  shortDeferRunAt: () => SHORT_DEFER_AT,
+}))
+
+const UNIT = {
+  id: 'unit-1',
+  integrationId: 'integration-1',
+  platform: 'github',
+  syncName: 'issues',
+  channelId: 'repo',
+  channelName: 'https://github.com/org/repo',
+  leaseLapsed: false,
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  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)
+    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, 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)
+    expect(mocks.rescheduleUnit).not.toHaveBeenCalled()
+  })
+})

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

@@ -1,12 +1,16 @@
 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'
 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'
@@ -40,8 +44,14 @@ export async function admitByBudget(units: IClaimedUnit[]): Promise<IAdmissionRe
   return result
 }
 
-export async function deferUnit(unitId: string): Promise<void> {
-  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<void> {
+  const qx = dbStoreQx(svc.postgres.writer)
+  if (leaseLapsed) {
+    await renewRunningUnitLease(qx, unitId)
+  } else {
+    await rescheduleUnit(qx, unitId, shortDeferRunAt())
+  }
 }
 
 function channelLabel(channelName: string): string {
@@ -86,18 +96,11 @@ export async function startRun(unit: IClaimedUnit): Promise<StartRunResult> {
 }
 
 async function backOffRunningUnit(unit: IClaimedUnit): Promise<void> {
-  const nextRunAt = await runningSince(unit)
-    .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')
-}
-
-async function runningSince(unit: IClaimedUnit): Promise<Date> {
-  const description = await svc.temporal.workflow.getHandle(syncRunWorkflowId(unit)).describe()
-
-  return description.startTime
+  const renewed = await renewRunningUnitLease(dbStoreQx(svc.postgres.writer), unit.id)
+  svc.log.info(
+    { unitId: unit.id, renewed: renewed > 0 },
+    'unit already running, lease renewed until it lapses',
+  )
 }
 
 export async function logDispatchSummary(counts: IDispatchCounts): Promise<void> {

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

@@ -0,0 +1,137 @@
+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(),
+  reassertUnitClaim: 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<typeof import('@crowd/logging')>()),
+  getChildLogger: () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }),
+}))
+
+vi.mock('@crowd/connectors', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@crowd/connectors')>()),
+  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,
+  reassertUnitClaim: mocks.reassertUnitClaim,
+  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<unknown>) => {
+    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,
+    )
+  })
+})
+
+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],
+    )
+  })
+})

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<void> {
   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`)
@@ -156,7 +158,10 @@ export async function executeSync(unitId: string): Promise<void> {
     }
 
     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,