fix: settle github activity worker queues

scripts/backfill-github-activity.ts

@@ -30,6 +30,8 @@ import { backfillGitHubCommitsFromCurrentRefs } from "../src/lib/github-direct-b
 import { backfillGitHubPullRequests } from "../src/lib/github-pull-request-backfill";
 
 const MAXIMUM_ACTION_MINUTES = 330;
+const MAXIMUM_INCONCLUSIVE_AUDIT_RETRIES = 2;
+const INCONCLUSIVE_AUDIT_RETRY_MS = 1000;
 const WORKER_CLEANUP_MARGIN_MS = 30_000;
 const WORKER_PASS_DURATION_MS = 240_000;
 
@@ -141,6 +143,8 @@ interface BackfillProcessingTotals extends GitHubBackfillProcessingCounts {
 
 const emptyProcessingCounts = (): GitHubBackfillProcessingCounts => ({
   aliases: 0,
+  canonicalizationAttempts: 0,
+  canonicalized: 0,
   claimed: 0,
   deferred: 0,
   failed: 0,
@@ -153,6 +157,8 @@ const addProcessingCounts = (
   counts: GitHubBackfillProcessingCounts
 ) => {
   totals.aliases += counts.aliases;
+  totals.canonicalizationAttempts += counts.canonicalizationAttempts;
+  totals.canonicalized += counts.canonicalized;
   totals.claimed += counts.claimed;
   totals.deferred += counts.deferred;
   totals.failed += counts.failed;
@@ -225,7 +231,7 @@ const processQueue = async (
     passes += 1;
     addProcessingCounts(totals, counts);
     process.stdout.write(
-      `Worker pass ${String(completedPasses + passes)}: claimed ${String(counts.claimed)}, processed ${String(counts.processed)}, deferred ${String(counts.deferred)}, unavailable ${String(counts.unavailable)}, failed ${String(counts.failed)}, aliases ${String(counts.aliases)}.\n`
+      `Worker pass ${String(completedPasses + passes)}: claimed ${String(counts.claimed)}, processed ${String(counts.processed)}, deferred ${String(counts.deferred)}, unavailable ${String(counts.unavailable)}, failed ${String(counts.failed)}, canonicalized ${String(counts.canonicalized)}/${String(counts.canonicalizationAttempts)}, aliases ${String(counts.aliases)}.\n`
     );
     if (!githubBackfillProcessingMadeProgress(counts)) {
       return {
@@ -251,18 +257,26 @@ const combineProcessingTotals = (
 export const backfillRetryWaitMillisecondsFrom = (
   audits: readonly Pick<GitHubActivityAuditReport, "pipeline" | "status">[],
   deadlineAt: number,
-  now = Date.now()
+  now = Date.now(),
+  inconclusiveRetries = 0
 ) => {
   if (
     audits.length === 0 ||
     audits.some(
       ({ status }) =>
+        status !== "inconclusive" &&
         status !== "pipeline_incomplete" &&
         status !== "stored_projection_verified"
     )
   ) {
     return null;
   }
+  if (audits.some(({ status }) => status === "inconclusive")) {
+    return inconclusiveRetries < MAXIMUM_INCONCLUSIVE_AUDIT_RETRIES &&
+      now + INCONCLUSIVE_AUDIT_RETRY_MS + WORKER_CLEANUP_MARGIN_MS < deadlineAt
+      ? INCONCLUSIVE_AUDIT_RETRY_MS
+      : null;
+  }
   const retryTimes = audits.flatMap(({ pipeline, status }) => {
     if (status !== "pipeline_incomplete" || pipeline.earliestRetryAt === null) {
       return [];
@@ -402,6 +416,7 @@ const main = async () => {
       stopReason: "no_immediately_claimable_work",
     };
     let audits: readonly GitHubActivityAuditReport[] = [];
+    let inconclusiveAuditRetries = 0;
     let completion = githubBackfillCompletionFrom({
       auditStatuses: [],
       boundedDiscoveryComplete,
@@ -421,7 +436,15 @@ const main = async () => {
         worker.stopReason = "time_budget";
         break;
       }
-      const retryWait = backfillRetryWaitMillisecondsFrom(audits, deadlineAt);
+      const hasInconclusiveAudit = audits.some(
+        ({ status }) => status === "inconclusive"
+      );
+      const retryWait = backfillRetryWaitMillisecondsFrom(
+        audits,
+        deadlineAt,
+        Date.now(),
+        inconclusiveAuditRetries
+      );
       if (retryWait === null) {
         if (
           audits.some(
@@ -438,6 +461,9 @@ const main = async () => {
         }
         break;
       }
+      inconclusiveAuditRetries = hasInconclusiveAudit
+        ? inconclusiveAuditRetries + 1
+        : 0;
       process.stdout.write(
         `No scoped work is claimable; waiting ${String(Math.ceil(retryWait / 1000))} seconds for the next durable retry within the Action budget.\n`
       );

src/app/api/cron/github-worker/route.ts

@@ -57,6 +57,7 @@ export async function POST(request: Request) {
       evidenceRecovery.status === "applied" ||
       activity.summaries.completed > 0 ||
       activity.pullRequests.completed > 0 ||
+      activity.canonicalizationAttempts > 0 ||
       activity.aliases > 0
     ) {
       revalidateTag("github-activity", "max");

src/lib/github-activity-audit.ts

@@ -15,6 +15,7 @@ import {
   githubIssues,
   githubPublicActivities,
   githubPullRequests,
+  githubPullRequestMemberships,
   githubPullRequestSignals,
   githubPullRequestVersions,
   githubPushObservations,
@@ -726,6 +727,9 @@ interface AuditRetryRow {
 interface AuditPullRequestPipelineRow {
   lastReconciledAt: Date | null;
   membershipComplete: boolean | null;
+  membershipCount: number;
+  membershipExpectedCount: number | null;
+  membershipHeadMatches: boolean;
   nextReconcileAt: Date | null;
   reconcileError: string | null;
   versionObservedAt: Date | null;
@@ -798,7 +802,11 @@ const pullRequestPipelineEvidenceFrom = (
   let pullRequestReconciliationsUnavailable = 0;
   const retryDates: Date[] = [];
   for (const pullRequest of rows) {
-    const membershipIncomplete = pullRequest.membershipComplete !== true;
+    const membershipIncomplete =
+      pullRequest.membershipComplete !== true ||
+      pullRequest.membershipExpectedCount === null ||
+      pullRequest.membershipCount !== pullRequest.membershipExpectedCount ||
+      !pullRequest.membershipHeadMatches;
     if (membershipIncomplete) {
       if (pullRequest.nextReconcileAt === null) {
         pullRequestMembershipsUnavailable += 1;
@@ -1083,6 +1091,22 @@ const readStoredEvidence = async (
       .select({
         lastReconciledAt: githubPullRequests.lastReconciledAt,
         membershipComplete: githubPullRequestVersions.membershipComplete,
+        membershipCount: sql<number>`(
+          SELECT count(*)::integer
+          FROM ${githubPullRequestMemberships}
+          WHERE ${githubPullRequestMemberships.versionId} = ${githubPullRequestVersions.id}
+        )`,
+        membershipExpectedCount: githubPullRequestVersions.commitCount,
+        membershipHeadMatches: sql<boolean>`
+          coalesce(${githubPullRequestVersions.commitCount} = 0, false)
+          OR coalesce((
+            SELECT ${githubPullRequestMemberships.commitSha} = ${githubPullRequestVersions.headSha}
+            FROM ${githubPullRequestMemberships}
+            WHERE ${githubPullRequestMemberships.versionId} = ${githubPullRequestVersions.id}
+            ORDER BY ${githubPullRequestMemberships.position} DESC
+            LIMIT 1
+          ), false)
+        `,
         nextReconcileAt: githubPullRequests.nextReconcileAt,
         reconcileError: githubPullRequests.reconcileError,
         versionObservedAt: githubPullRequestVersions.observedAt,

src/lib/github-activity-worker-store.ts

@@ -577,7 +577,10 @@ export const claimGitHubPushObservations = async (
         beforeSha: githubPushObservations.beforeSha,
         errorCode: githubPushObservations.errorCode,
         expectedCommitCount: githubPushObservations.expectedCommitCount,
-        historySinceAt: sql<Date>`coalesce(${githubPushObservations.historySinceAt}, ${githubAccountCheckpoints.refBackfillSinceAt})`,
+        historySinceAt:
+          sql`coalesce(${githubPushObservations.historySinceAt}, ${githubAccountCheckpoints.refBackfillSinceAt})`.mapWith(
+            githubAccountCheckpoints.refBackfillSinceAt
+          ),
         historyUntilAt: githubPushObservations.historyUntilAt,
         id: githubPushObservations.id,
         leaseUntil: githubPushObservations.leaseUntil,
@@ -1738,6 +1741,18 @@ const persistPullRequestSnapshotInTransaction = async (
       id: githubPullRequestVersions.id,
       isCurrent: githubPullRequestVersions.isCurrent,
       membershipComplete: githubPullRequestVersions.membershipComplete,
+      membershipCount: sql<number>`(
+        SELECT count(*)::integer
+        FROM ${githubPullRequestMemberships}
+        WHERE ${githubPullRequestMemberships.versionId} = ${githubPullRequestVersions.id}
+      )`,
+      membershipHeadSha: sql<string | null>`(
+        SELECT ${githubPullRequestMemberships.commitSha}
+        FROM ${githubPullRequestMemberships}
+        WHERE ${githubPullRequestMemberships.versionId} = ${githubPullRequestVersions.id}
+        ORDER BY ${githubPullRequestMemberships.position} DESC
+        LIMIT 1
+      )`,
       providerUpdatedAt: githubPullRequestVersions.providerUpdatedAt,
     })
     .from(githubPullRequestVersions)
@@ -1805,16 +1820,43 @@ const persistPullRequestSnapshotInTransaction = async (
           pullRequest.headRepository?.id ?? version.headRepositoryId,
         isCurrent: true,
         mergeSnapshot: pullRequest.merged,
-        observedAt: now,
+        ...(providerUpdatedAt > version.providerUpdatedAt ||
+        canonicalEvidenceChanged
+          ? { observedAt: now }
+          : {}),
         providerUpdatedAt,
       })
       .where(eq(githubPullRequestVersions.id, version.id));
   }
   if (versionId === undefined) {
     throw new Error("The GitHub pull request version is unavailable.");
   }
+  const expectedMembershipCount =
+    pullRequest.commitCount ?? version?.commitCount ?? null;
+  const storedMembershipCompleteFlag = version?.membershipComplete ?? false;
+  const storedMembershipComplete =
+    storedMembershipCompleteFlag &&
+    expectedMembershipCount !== null &&
+    version.membershipCount === expectedMembershipCount &&
+    (expectedMembershipCount === 0
+      ? version.membershipHeadSha === null
+      : version.membershipHeadSha === pullRequest.headSha);
+  const storedMembershipInvalid =
+    storedMembershipCompleteFlag && !storedMembershipComplete;
+  if (storedMembershipInvalid) {
+    await invalidateGitHubPullRequestDerivedAliases(
+      transaction,
+      pullRequest.nodeId
+    );
+    await transaction
+      .update(githubPullRequestVersions)
+      .set({ membershipComplete: false })
+      .where(eq(githubPullRequestVersions.id, versionId));
+  }
   const membershipRefreshRequired =
-    version === undefined || (refreshMembership && !version.membershipComplete);
+    version === undefined ||
+    storedMembershipInvalid ||
+    (refreshMembership && !storedMembershipComplete);
   if (membershipRefreshRequired && existing !== undefined) {
     await transaction
       .update(githubPullRequests)
@@ -2048,14 +2090,29 @@ export const persistGitHubPullRequestMembership = async (
 ) =>
   await getDatabase().transaction(async (transaction) => {
     const [version] = await transaction
-      .select({ id: githubPullRequestVersions.id })
+      .select({
+        commitCount: githubPullRequestVersions.commitCount,
+        headSha: githubPullRequestVersions.headSha,
+        id: githubPullRequestVersions.id,
+      })
       .from(githubPullRequestVersions)
       .where(eq(githubPullRequestVersions.id, stored.versionId))
       .for("update");
     if (version === undefined) {
       return false;
     }
-    if (!membershipComplete) {
+    const completeMembershipIsValid =
+      membershipComplete &&
+      version.commitCount !== null &&
+      commitShas.length === version.commitCount &&
+      new Set(commitShas).size === commitShas.length &&
+      headSha === version.headSha &&
+      (version.commitCount === 0 || commitShas.at(-1) === version.headSha);
+    if (!completeMembershipIsValid) {
+      await invalidateGitHubPullRequestDerivedAliases(
+        transaction,
+        stored.pullRequestNodeId
+      );
       await transaction
         .update(githubPullRequestVersions)
         .set({ membershipComplete: false })
@@ -2398,7 +2455,8 @@ const markGitHubCommitCanonicalized = async (
 export const canonicalizeGitHubCommitActivity = async (
   repositoryId: string,
   sha: string,
-  now = new Date()
+  now = new Date(),
+  options: { onlyIfPending?: boolean } = {}
 ) =>
   // oxlint-disable-next-line complexity -- Evidence classes deliberately fail open independently.
   await getDatabase().transaction(async (transaction) => {
@@ -2426,6 +2484,9 @@ export const canonicalizeGitHubCommitActivity = async (
             "complete",
             "unavailable",
           ]),
+          options.onlyIfPending === true
+            ? isNull(githubCommits.canonicalizedAt)
+            : undefined,
           isNull(githubPublicActivities.canonicalPublicId),
           isNull(githubPublicActivities.hiddenAt)
         )
@@ -2641,7 +2702,7 @@ export const canonicalizePendingGitHubActivities = async (
   scope?: GitHubActivityWorkerScope
 ) => {
   if (activeAccounts.length === 0) {
-    return 0;
+    return { aliases: 0, canonicalizationAttempts: 0, canonicalized: 0 };
   }
   const candidates = await getDatabase()
     .select({
@@ -2667,15 +2728,22 @@ export const canonicalizePendingGitHubActivities = async (
     .orderBy(asc(githubCommits.firstObservedAt), asc(githubCommits.sha))
     .limit(limit);
   let aliased = 0;
+  let canonicalized = 0;
   for (const candidate of candidates) {
     const result = await canonicalizeGitHubCommitActivity(
       candidate.repositoryId,
       candidate.sha,
-      now
+      now,
+      { onlyIfPending: true }
     );
     aliased += result.aliases;
+    canonicalized += result.publicId === null ? 0 : 1;
   }
-  return aliased;
+  return {
+    aliases: aliased,
+    canonicalizationAttempts: candidates.length,
+    canonicalized,
+  };
 };
 
 export const ensureGitHubSummaryAttempt = async (

src/lib/github-activity-worker.ts

@@ -100,6 +100,8 @@ const emptyStageResult = (): StageResult => ({
 
 export interface GitHubActivityWorkerResult {
   aliases: number;
+  canonicalizationAttempts: number;
+  canonicalized: number;
   commits: StageResult;
   deadlineReached: boolean;
   observations: StageResult;
@@ -677,8 +679,8 @@ export const runGitHubActivityWorker = async (
     pullRequestDiscovery
   );
   await processPullRequests(context, pullRequestLimit, pullRequests);
-  const aliases = context.deadlineReached()
-    ? 0
+  const canonicalization = context.deadlineReached()
+    ? { aliases: 0, canonicalizationAttempts: 0, canonicalized: 0 }
     : await canonicalizePendingGitHubActivities(
         8,
         new Date(),
@@ -696,7 +698,9 @@ export const runGitHubActivityWorker = async (
   await processSummaries(context, summaryLimit, summaries);
 
   return {
-    aliases,
+    aliases: canonicalization.aliases,
+    canonicalizationAttempts: canonicalization.canonicalizationAttempts,
+    canonicalized: canonicalization.canonicalized,
     commits,
     deadlineReached: context.deadlineReached(),
     observations,

src/lib/github-backfill-core.ts

@@ -80,6 +80,8 @@ interface GitHubBackfillWorkerStageCounts {
 
 interface GitHubBackfillWorkerPassCounts {
   aliases: number;
+  canonicalizationAttempts: number;
+  canonicalized: number;
   commits: GitHubBackfillWorkerStageCounts;
   observations: GitHubBackfillWorkerStageCounts;
   pullRequestDiscovery: GitHubBackfillWorkerStageCounts;
@@ -90,6 +92,8 @@ interface GitHubBackfillWorkerPassCounts {
 
 export interface GitHubBackfillProcessingCounts {
   aliases: number;
+  canonicalizationAttempts: number;
+  canonicalized: number;
   claimed: number;
   deferred: number;
   failed: number;
@@ -98,8 +102,11 @@ export interface GitHubBackfillProcessingCounts {
 }
 
 export const githubBackfillProcessingMadeProgress = (
-  counts: Pick<GitHubBackfillProcessingCounts, "aliases" | "claimed">
-) => counts.claimed > 0 || counts.aliases > 0;
+  counts: Pick<
+    GitHubBackfillProcessingCounts,
+    "canonicalizationAttempts" | "claimed"
+  >
+) => counts.claimed > 0 || counts.canonicalizationAttempts > 0;
 
 type JsonObject = Record<string, unknown>;
 
@@ -111,6 +118,8 @@ export const githubBackfillProcessingCountsFrom = (
 ): GitHubBackfillProcessingCounts => {
   const counts: GitHubBackfillProcessingCounts = {
     aliases: result.aliases,
+    canonicalizationAttempts: result.canonicalizationAttempts,
+    canonicalized: result.canonicalized,
     claimed: 0,
     deferred: 0,
     failed: 0,