fix: retry rejected capture batches and self-heal newly-gapped repos (CM-1441)

services/apps/cron_service/src/jobs/starSnapshotHealthReporting.job.ts

@@ -4,8 +4,8 @@ import { IS_DEV_ENV, IS_PROD_ENV } from '@crowd/common'
 import {
   IRepoStarSnapshotGapDays,
   countDeadLetteredStarBackfillFailures,
+  findAllRepoIdsWithStarSnapshotGaps,
   findDeadLetteredStarBackfillFailures,
-  findRepoIdsWithStarSnapshotGaps,
   findReposForStarSnapshot,
   findStarSnapshotGapDaysForRepos,
   getDeadLetterReportCursor,
@@ -25,8 +25,7 @@ import { IJobDefinition } from '../types'
 const LAST_DEAD_LETTER_REPORTED_AT_KEY =
   'star-snapshot-health-reporting:last-dead-letter-reported-at'
 const SAMPLE_SIZE = 20
-// Keeps each gap-check query's IN-list bounded as the eligible repo count grows.
-const GAP_CHECK_BATCH_SIZE = 5_000
+const GAP_DAYS_BATCH_SIZE = 5_000
 
 const job: IJobDefinition = {
   name: 'star-snapshot-health-reporting',
@@ -55,27 +54,28 @@ const job: IJobDefinition = {
 
     const repoUrlById = new Map(allRepos.map((repo) => [repo.repositoryId, repo.repoUrl]))
     const allRepoIds = allRepos.map((repo) => repo.repositoryId)
-    const gappedRepoIds: string[] = []
-    for (let i = 0; i < allRepoIds.length; i += GAP_CHECK_BATCH_SIZE) {
-      const batch = allRepoIds.slice(i, i + GAP_CHECK_BATCH_SIZE)
-      gappedRepoIds.push(...(await findRepoIdsWithStarSnapshotGaps(qx, batch)))
-    }
+    const gappedRepoIds = await findAllRepoIdsWithStarSnapshotGaps(qx, allRepoIds)
 
     const gapDays: IRepoStarSnapshotGapDays[] = []
-    for (let i = 0; i < gappedRepoIds.length; i += GAP_CHECK_BATCH_SIZE) {
-      const batch = gappedRepoIds.slice(i, i + GAP_CHECK_BATCH_SIZE)
+    for (let i = 0; i < gappedRepoIds.length; i += GAP_DAYS_BATCH_SIZE) {
+      const batch = gappedRepoIds.slice(i, i + GAP_DAYS_BATCH_SIZE)
       gapDays.push(...(await findStarSnapshotGapDaysForRepos(qx, batch)))
     }
     const missingDaysByRepoId = new Map(gapDays.map((gap) => [gap.repositoryId, gap.missingDays]))
     const totalMissingDays = gapDays.reduce((sum, gap) => sum + gap.missingDays, 0)
+    // The two queries above run seconds apart, so a repo can close its gap in between and
+    // come back with 0 missing days - drop those instead of over-reporting the gap count.
+    const currentlyGappedRepoIds = gapDays
+      .filter((gap) => gap.missingDays > 0)
+      .map((gap) => gap.repositoryId)
 
     const sections: SlackMessageSection[] = [
       {
         title: 'Star Snapshot Health Summary',
         text: [
           `🪦 New repos GitHub gave up retrying (3 failures in a row, excl. repo-gone/IP-allowlist): *${newlyDeadLettered.length}*`,
           `📉 Total repos GitHub gave up retrying: *${totalDeadLettered}*`,
-          `📅 Repos with a snapshot gap right now: *${gappedRepoIds.length}*`,
+          `📅 Repos with a snapshot gap right now: *${currentlyGappedRepoIds.length}*`,
           `📆 Total missing snapshot-days across those repos: *${totalMissingDays}*`,
         ].join('\n'),
       },
@@ -93,8 +93,8 @@ const job: IJobDefinition = {
       })
     }
 
-    if (gappedRepoIds.length > 0) {
-      const sortedByMissingDays = [...gappedRepoIds].sort(
+    if (currentlyGappedRepoIds.length > 0) {
+      const sortedByMissingDays = [...currentlyGappedRepoIds].sort(
         (a, b) => (missingDaysByRepoId.get(b) ?? 0) - (missingDaysByRepoId.get(a) ?? 0),
       )
       const shown = sortedByMissingDays.slice(0, SAMPLE_SIZE)
@@ -104,13 +104,13 @@ const job: IJobDefinition = {
         return `• \`${url}\` - missing ${days} day${days === 1 ? '' : 's'}`
       })
       sections.push({
-        title: `Snapshot Gaps (top ${shown.length} of ${gappedRepoIds.length}, most days missing first)`,
+        title: `Snapshot Gaps (top ${shown.length} of ${currentlyGappedRepoIds.length}, most days missing first)`,
         text: lines.join('\n'),
       })
     }
 
     const persona =
-      newlyDeadLettered.length > 0 || gappedRepoIds.length > 0
+      newlyDeadLettered.length > 0 || currentlyGappedRepoIds.length > 0
         ? SlackPersona.WARNING_PROPAGATOR
         : SlackPersona.INFO_NOTIFIER
 
@@ -128,7 +128,7 @@ const job: IJobDefinition = {
     }
 
     ctx.log.info(
-      `Star snapshot health report sent: newlyDeadLettered=${newlyDeadLettered.length}, totalDeadLettered=${totalDeadLettered}, gaps=${gappedRepoIds.length}`,
+      `Star snapshot health report sent: newlyDeadLettered=${newlyDeadLettered.length}, totalDeadLettered=${totalDeadLettered}, gaps=${currentlyGappedRepoIds.length}`,
     )
   },
 }

services/apps/star_snapshot_worker/src/activities.ts

@@ -2,12 +2,14 @@ import {
   backfillRepoStarHistory,
   fetchAndSaveStarSnapshotBatch,
   findReposForStarSnapshot,
+  findReposNeedingGapHeal,
   findReposNeedingStarBackfill,
 } from './activities/index'
 
 export {
   backfillRepoStarHistory,
   fetchAndSaveStarSnapshotBatch,
   findReposForStarSnapshot,
+  findReposNeedingGapHeal,
   findReposNeedingStarBackfill,
 }

services/apps/star_snapshot_worker/src/activities/index.ts

@@ -2,6 +2,8 @@ import { ApplicationFailure } from '@temporalio/client'
 
 import { getGithubInstallationToken } from '@crowd/common_services'
 import {
+  findAllRepoIdsWithStarSnapshotGaps,
+  findCompletedReposEligibleForGapHeal,
   findReposForStarSnapshot as findReposForStarSnapshotQx,
   findReposNeedingStarBackfill as findReposNeedingStarBackfillQx,
   recordStarBackfillFailure,
@@ -395,6 +397,33 @@ export async function findReposNeedingStarBackfill(
   return findReposNeedingStarBackfillQx(qx, limit, afterUrl)
 }
 
+export interface IGapHealPage {
+  gappedRepos: IRepoForStarSnapshot[]
+  pageSize: number
+  lastUrl?: string
+}
+
+// A completed repo can still pick up a fresh gap (e.g. a dropped capture batch) - this
+// finds those so selfHealStarBackfill re-sweeps them too, paginated like the backfill scan.
+export async function findReposNeedingGapHeal(
+  limit: number,
+  afterUrl?: string,
+): Promise<IGapHealPage> {
+  const qx = pgpQx(svc.postgres.reader.connection())
+  const page = await findCompletedReposEligibleForGapHeal(qx, limit, afterUrl)
+  const gappedIds = new Set(
+    await findAllRepoIdsWithStarSnapshotGaps(
+      qx,
+      page.map((repo) => repo.repositoryId),
+    ),
+  )
+  return {
+    gappedRepos: page.filter((repo) => gappedIds.has(repo.repositoryId)),
+    pageSize: page.length,
+    lastUrl: page.length > 0 ? page[page.length - 1].repoUrl : undefined,
+  }
+}
+
 // A leftover claim only costs a repo one skipped run before the TTL clears it - not worth
 // forcing a full activity retry (re-fetching the entire stargazer history) over.
 async function releaseInflightClaim(cache: RedisCache, repositoryId: string): Promise<void> {

services/apps/star_snapshot_worker/src/bin/star-snapshot-backfill.ts

@@ -2,7 +2,11 @@ import { randomUUID } from 'crypto'
 import { mkdir, readFile, rename, rm, writeFile } from 'fs/promises'
 import { dirname } from 'path'
 
-import { findRepoIdsWithStarSnapshotGaps } from '@crowd/data-access-layer'
+import {
+  findAllRepoIdsWithStarSnapshotGaps,
+  findRepoIdsWithStarSnapshotGaps,
+  findReposForStarSnapshot,
+} from '@crowd/data-access-layer'
 import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { getServiceLogger } from '@crowd/logging'
@@ -29,6 +33,12 @@ const DEFAULT_RESERVED_CORE_RATE_LIMIT = 2_000
 const DEFAULT_CONCURRENCY = 5
 const DEFAULT_CHECKPOINT_FILE = '/var/lib/star-snapshot-worker/backfill-checkpoint.json'
 const DEFAULT_COMPLETED_REPOS_FILE = '/var/lib/star-snapshot-worker/backfill-completed-repos.json'
+// --gapped-only keeps its own progress files so it can't clobber a normal full-sweep run's
+// checkpoint/completed-repos state (or vice versa) if the two are ever run against each other.
+const DEFAULT_GAPPED_ONLY_CHECKPOINT_FILE =
+  '/var/lib/star-snapshot-worker/backfill-gapped-only-checkpoint.json'
+const DEFAULT_GAPPED_ONLY_COMPLETED_REPOS_FILE =
+  '/var/lib/star-snapshot-worker/backfill-gapped-only-completed-repos.json'
 
 interface Checkpoint {
   afterUrl: string
@@ -106,11 +116,18 @@ async function clearCompletedRepoIds(path: string): Promise<void> {
 const main = async () => {
   const dryRun = process.argv.includes('--dry-run')
   const fresh = process.argv.includes('--fresh')
+  // Targets only currently-gapped repos instead of sweeping every eligible one - fixes a
+  // specific gap on demand without a full backfill pass.
+  const gappedOnly = process.argv.includes('--gapped-only')
   const afterUrlOverride = readFlagValue('--after-url')
-  const checkpointFile =
-    process.env.STAR_SNAPSHOT_BACKFILL_CHECKPOINT_FILE ?? DEFAULT_CHECKPOINT_FILE
-  const completedReposFile =
-    process.env.STAR_SNAPSHOT_BACKFILL_COMPLETED_REPOS_FILE ?? DEFAULT_COMPLETED_REPOS_FILE
+  const checkpointFile = gappedOnly
+    ? (process.env.STAR_SNAPSHOT_BACKFILL_GAPPED_ONLY_CHECKPOINT_FILE ??
+      DEFAULT_GAPPED_ONLY_CHECKPOINT_FILE)
+    : (process.env.STAR_SNAPSHOT_BACKFILL_CHECKPOINT_FILE ?? DEFAULT_CHECKPOINT_FILE)
+  const completedReposFile = gappedOnly
+    ? (process.env.STAR_SNAPSHOT_BACKFILL_GAPPED_ONLY_COMPLETED_REPOS_FILE ??
+      DEFAULT_GAPPED_ONLY_COMPLETED_REPOS_FILE)
+    : (process.env.STAR_SNAPSHOT_BACKFILL_COMPLETED_REPOS_FILE ?? DEFAULT_COMPLETED_REPOS_FILE)
   const reservedCoreRateLimit = readIntEnv(
     'STAR_SNAPSHOT_BACKFILL_RESERVED_CORE_RATE_LIMIT',
     DEFAULT_RESERVED_CORE_RATE_LIMIT,
@@ -119,12 +136,14 @@ const main = async () => {
   const concurrency = readIntEnv('STAR_SNAPSHOT_BACKFILL_CONCURRENCY', DEFAULT_CONCURRENCY, false)
 
   let afterUrl = afterUrlOverride
+  let resumedFromCheckpoint = false
   if (!afterUrl && fresh) {
     await clearCheckpoint(checkpointFile)
   } else if (!afterUrl) {
     const checkpoint = await readCheckpoint(checkpointFile)
     if (checkpoint) {
       afterUrl = checkpoint.afterUrl
+      resumedFromCheckpoint = true
       log.info({ checkpoint }, 'resuming star snapshot backfill from checkpoint')
     }
   }
@@ -142,6 +161,7 @@ const main = async () => {
     {
       dryRun,
       fresh,
+      gappedOnly,
       afterUrl,
       checkpointFile,
       completedReposFile,
@@ -157,9 +177,31 @@ const main = async () => {
   await qx.selectOne('SELECT 1')
   log.info('Connected to database.')
 
-  // A repo can be marked complete yet still have a gap (e.g. an earlier interrupted
-  // run) - re-check and reprocess it instead of trusting the flag forever.
-  if (completedRepoIds.size > 0) {
+  if (gappedOnly && !resumedFromCheckpoint) {
+    // Everyone not currently gapped is pre-marked "completed" so the scan/skip loop below
+    // flies through them, touching only the repos actually gapped right now.
+    const allRepos = await findReposForStarSnapshot(qx)
+    const gappedRepoIds = new Set(
+      await findAllRepoIdsWithStarSnapshotGaps(
+        qx,
+        allRepos.map((repo) => repo.repositoryId),
+      ),
+    )
+    completedRepoIds.clear()
+    for (const repo of allRepos) {
+      if (!gappedRepoIds.has(repo.repositoryId)) {
+        completedRepoIds.add(repo.repositoryId)
+      }
+    }
+    log.info(
+      { totalRepos: allRepos.length, gappedRepoCount: gappedRepoIds.size },
+      'gapped-only backfill targeting currently-gapped repos',
+    )
+  }
+
+  // A repo can be marked complete yet still have a gap (e.g. an interrupted run) - re-check
+  // it. Skipped in gapped-only mode, which already re-derives completedRepoIds every run.
+  if (!gappedOnly && completedRepoIds.size > 0) {
     const gappedRepoIds = await findRepoIdsWithStarSnapshotGaps(qx, [...completedRepoIds])
     const unmarked = gappedRepoIds.filter((id) => completedRepoIds.delete(id)).length
     if (unmarked > 0) {

services/apps/star_snapshot_worker/src/workflows/captureStarSnapshots.ts

@@ -2,6 +2,7 @@ import {
   ApplicationFailure,
   continueAsNew,
   log,
+  patched,
   proxyActivities,
   sleep,
   workflowInfo,
@@ -19,6 +20,9 @@ const { findReposForStarSnapshot, fetchAndSaveStarSnapshotBatch } = proxyActivit
 const GRAPHQL_BATCH_SIZE = 100
 const CONCURRENCY = 5
 const PAGE_SIZE = 2_000
+// A rejected batch usually means a transient blip - one retry pass after a cooldown
+// recovers most of them instead of leaving a permanent gap for that day.
+const REJECTED_BATCH_RETRY_DELAY_MS = 30_000
 
 export interface ICaptureStarSnapshotsArgs {
   capturedAt?: string
@@ -39,12 +43,15 @@ export async function captureStarSnapshots(args: ICaptureStarSnapshotsArgs = {})
 
   let succeeded = args.succeededSoFar ?? 0
   let failed = args.failedSoFar ?? 0
-  let rejectedBatches = 0
+  let rejectedBatches: (typeof batches)[number][] = []
 
-  for (let i = 0; i < batches.length; i += CONCURRENCY) {
-    // Rate-limited batches retry in place via a durable workflow sleep, not a blocking
-    // activity call - mirrors backfillStarHistoryBatch's handling of the same quota problem.
-    let window = batches.slice(i, i + CONCURRENCY)
+  // Runs one window of batches to completion, handling rate-limit backoff in place.
+  // Returns the batches that were still rejected (activity retries exhausted) when done.
+  const runWindow = async (
+    initialWindow: (typeof batches)[number][],
+  ): Promise<(typeof batches)[number][]> => {
+    let window = initialWindow
+    const rejected: (typeof batches)[number][] = []
 
     while (window.length > 0) {
       const results = await Promise.allSettled(
@@ -56,8 +63,7 @@ export async function captureStarSnapshots(args: ICaptureStarSnapshotsArgs = {})
 
       for (const [idx, result] of results.entries()) {
         if (result.status === 'rejected') {
-          rejectedBatches++
-          failed += window[idx].length
+          rejected.push(window[idx])
           log.warn('Failed to capture star snapshot batch', {
             repoCount: window[idx].length,
             error: (result.reason as Error)?.message ?? result.reason,
@@ -93,13 +99,37 @@ export async function captureStarSnapshots(args: ICaptureStarSnapshotsArgs = {})
         await sleep(waitMs)
       }
     }
+
+    return rejected
+  }
+
+  for (let i = 0; i < batches.length; i += CONCURRENCY) {
+    rejectedBatches.push(...(await runWindow(batches.slice(i, i + CONCURRENCY))))
+  }
+
+  // patched() keeps an execution already in flight on its old command sequence so a
+  // mid-deploy replay doesn't hit a nondeterminism error.
+  if (rejectedBatches.length > 0 && patched('CM-1441-retry-rejected-batches')) {
+    await sleep(REJECTED_BATCH_RETRY_DELAY_MS)
+    const stillRejected: (typeof batches)[number][] = []
+    for (let i = 0; i < rejectedBatches.length; i += CONCURRENCY) {
+      stillRejected.push(...(await runWindow(rejectedBatches.slice(i, i + CONCURRENCY))))
+    }
+    for (const batch of stillRejected) {
+      failed += batch.length
+    }
+    rejectedBatches = stillRejected
+  } else {
+    for (const batch of rejectedBatches) {
+      failed += batch.length
+    }
   }
 
   const total = (args.totalSoFar ?? 0) + repos.length
 
   // A few rejected batches self-heal (next run's diff, or the gap backfill) - only a fully
   // wiped-out page signals something systemic (auth/token/outage) worth failing the run over.
-  if (batches.length > 0 && rejectedBatches === batches.length) {
+  if (batches.length > 0 && rejectedBatches.length === batches.length) {
     // A plain Error only fails the workflow task (infinite replay); ApplicationFailure
     // is required to fail the execution so the schedule's retry policy engages.
     throw ApplicationFailure.create({
@@ -108,9 +138,9 @@ export async function captureStarSnapshots(args: ICaptureStarSnapshotsArgs = {})
     })
   }
 
-  if (rejectedBatches > 0) {
+  if (rejectedBatches.length > 0) {
     log.error('star snapshot capture had partial batch failures on this page', {
-      rejectedBatches,
+      rejectedBatches: rejectedBatches.length,
       totalBatches: batches.length,
       succeeded,
       failed,