fix: add automatic queue self-healing and retry unblocking (#97)

* fix: add automatic queue self-healing

* fix: avoid queue starvation during retries

* fix: requeue stale active jobs without builds

Author: frostebite

functions/src/cron/index.ts

@@ -17,12 +17,12 @@ if (MINUTES < 10) {
   throw new Error('Is the result really worth the machine time? Remove me.');
 }
 
-// Timeout of 60 seconds will keep our routine process tight.
+// Allow enough time for DockerHub-backed healing work during queue recovery.
 export const trigger = onSchedule(
   {
     schedule: `every ${MINUTES} minutes`,
     memory: '512MiB',
-    timeoutSeconds: 60,
+    timeoutSeconds: 300,
     secrets: [discordToken, githubPrivateKeyConfigSecret, githubClientSecretConfigSecret],
   },
   async () => {

functions/src/logic/buildQueue/cleanUpBuilds.ts

@@ -1,5 +1,7 @@
 import { Cleaner } from './cleaner';
+import { RepoVersionInfo } from '../../model/repoVersionInfo';
 
 export const cleanUpBuilds = async () => {
-  await Cleaner.cleanUp();
+  const latestRepoVersion = await RepoVersionInfo.getLatest();
+  await Cleaner.cleanUp(latestRepoVersion.version);
 };

functions/src/logic/buildQueue/cleaner.ts

@@ -1,15 +1,23 @@
 import { CiBuilds } from '../../model/ciBuilds';
+import { CiJobs } from '../../model/ciJobs';
 import { Discord } from '../../service/discord';
 import { Dockerhub } from '../../service/dockerhub';
 
 export class Cleaner {
-  // Cronjob intentionally has a limited runtime
-  static readonly maxBuildsProcessedPerRun: number = 5;
+  // Cronjob intentionally has a limited runtime, but it still needs enough
+  // throughput to drain queue backlogs without requiring manual intervention.
+  static readonly maxBuildsProcessedPerRun: number = 25;
+  static readonly activeJobWithoutBuildsAfterMinutes: number = 30;
+  static readonly startedBuildPublishProbeAfterMinutes: number = 45;
+  static readonly startedBuildFailureAfterHours: number = 6;
 
   static buildsProcessed: number;
 
-  public static async cleanUp() {
+  public static async cleanUp(latestRepoVersion: string) {
     this.buildsProcessed = 0;
+    await this.requeueActiveJobsWithoutBuilds();
+    await this.recoverMaxedOutFailedBuilds(latestRepoVersion);
+    await this.reconcileStartedBuildsThatMayHavePublished();
     await this.cleanUpBuildsThatDidntReportBack();
     await this.healFailedBuildsAlreadyOnDockerHub();
   }
@@ -22,6 +30,36 @@ export class Cleaner {
    */
   static readonly maxRecoveryAttempts: number = 2;
 
+  /**
+   * Jobs can get stuck in "scheduled" or "inProgress" without ever creating a
+   * build record when the dispatch path flakes before reportNewBuild. Reset
+   * those jobs back to created so the scheduler can dispatch them again.
+   */
+  private static async requeueActiveJobsWithoutBuilds() {
+    const activeJobs = await CiJobs.getActiveJobs();
+    const staleThresholdMs = this.activeJobWithoutBuildsAfterMinutes * 60 * 1000;
+
+    for (const activeJob of activeJobs) {
+      if (this.buildsProcessed >= this.maxBuildsProcessedPerRun) return;
+
+      const { id: jobId, data: job } = activeJob;
+      const lastTouchedSeconds = job.modifiedDate?.seconds || job.addedDate?.seconds;
+      if (!lastTouchedSeconds) continue;
+
+      const ageMs = Date.now() - lastTouchedSeconds * 1000;
+      if (ageMs < staleThresholdMs) continue;
+
+      const hasBuilds = await CiBuilds.hasAnyBuildsForJob(jobId);
+      if (hasBuilds) continue;
+
+      this.buildsProcessed += 1;
+      await CiJobs.resetJobToCreated(jobId);
+      await Discord.sendAlert(
+        `[Cleaner] Reset stale ${job.status} job "${jobId}" back to created because it had no build records after ${this.activeJobWithoutBuildsAfterMinutes} minutes.`,
+      );
+    }
+  }
+
   /**
    * Automatically recover maxed-out failed builds for the latest repo version.
    * If the image is already on DockerHub, mark it as published.
@@ -183,6 +221,49 @@ export class Cleaner {
     }
   }
 
+  /**
+   * If a started build has been running for a while and the image already
+   * exists on DockerHub, we can mark it as published without waiting for the
+   * full GitHub Actions timeout window. This frees queue capacity earlier.
+   */
+  private static async reconcileStartedBuildsThatMayHavePublished() {
+    const startedBuilds = await CiBuilds.getStartedBuilds();
+    const probeThresholdMs = this.startedBuildPublishProbeAfterMinutes * 60 * 1000;
+
+    for (const startedBuild of startedBuilds) {
+      if (this.buildsProcessed >= this.maxBuildsProcessedPerRun) return;
+
+      const { buildId, meta, relatedJobId: jobId, imageType, buildInfo } = startedBuild;
+      const { lastBuildStart, publishedDate } = meta;
+      const { baseOs, repoVersion } = buildInfo;
+
+      if (!lastBuildStart) continue;
+
+      const buildStartMs = new Date(lastBuildStart.seconds * 1000).getTime();
+      const nowMs = Date.now();
+      if (nowMs - buildStartMs < probeThresholdMs) continue;
+
+      const tag = buildId.replace(new RegExp(`^${imageType}-`), '');
+      this.buildsProcessed += 1;
+
+      const response = await Dockerhub.fetchImageData(imageType, tag);
+      if (!response) continue;
+
+      const digest = response.digest || '';
+      const message = publishedDate
+        ? `[Cleaner] Build "${tag}" has published metadata and exists on DockerHub. Reconciling status back to published.`
+        : `[Cleaner] Build "${tag}" is still "started" but already exists on DockerHub. Marking as published early.`;
+      await Discord.sendDebug(message);
+      await CiBuilds.markBuildAsPublished(buildId, jobId, {
+        digest,
+        specificTag: `${baseOs}-${repoVersion}`,
+        friendlyTag: repoVersion.replace(/\.\d+$/, ''),
+        imageName: Dockerhub.getImageName(imageType),
+        imageRepo: Dockerhub.getRepositoryBaseName(),
+      });
+    }
+  }
+
   private static async cleanUpBuildsThatDidntReportBack() {
     const startedBuilds = await CiBuilds.getStartedBuilds();
 
@@ -218,7 +299,8 @@ export class Cleaner {
       // If a job reaches this limit, the job is terminated and fails to complete.
       // @see https://docs.github.com/en/actions/learn-github-actions/usage-limits-billing-and-administration
       const ONE_HOUR = 1000 * 60 * 60;
-      const sixHoursAgo = new Date().getTime() - 6 * ONE_HOUR;
+      const failureThresholdMs = this.startedBuildFailureAfterHours * ONE_HOUR;
+      const sixHoursAgo = new Date().getTime() - failureThresholdMs;
       const buildStart = new Date(lastBuildStart.seconds * 1000).getTime();
 
       if (buildStart < sixHoursAgo) {

functions/src/logic/buildQueue/ingeminator.ts

@@ -14,13 +14,14 @@ import { logger } from 'firebase-functions/v2';
 export class Ingeminator {
   numberToSchedule: number;
   gitHubClient: Octokit;
+  private scheduledBuilds = 0;
 
   constructor(numberToSchedule: number, gitHubClient: Octokit) {
     this.numberToSchedule = numberToSchedule;
     this.gitHubClient = gitHubClient;
   }
 
-  async rescheduleFailedJobs(jobs: CiJobQueue) {
+  async rescheduleFailedJobs(jobs: CiJobQueue): Promise<number> {
     if (jobs.length <= 0) {
       throw new Error(
         '[Ingeminator] Expected ingeminator to be called with jobs to retry, none were given.',
@@ -36,6 +37,8 @@ export class Ingeminator {
 
       await this.rescheduleFailedBuildsForJob(job);
     }
+
+    return this.scheduledBuilds;
   }
 
   private async rescheduleFailedBuildsForJob(job: CiJobQueueItem) {
@@ -101,6 +104,7 @@ export class Ingeminator {
       if (!(await this.rescheduleBuild(jobId, jobData, buildId, BuildData))) {
         return;
       }
+      this.scheduledBuilds += 1;
     }
 
     await CiJobs.markJobAsScheduled(jobId);

functions/src/logic/buildQueue/scheduler.ts

@@ -19,6 +19,7 @@ export class Scheduler {
   private _gitHub: Octokit | undefined;
   private maxConcurrentJobs: number;
   private repoVersionInfo: RepoVersionInfo;
+  private reservedRetrySlots = 0;
 
   private get gitHub(): Octokit {
     // @ts-ignore
@@ -166,6 +167,7 @@ export class Scheduler {
   async ensureThereAreNoFailedJobs(): Promise<boolean> {
     const { maxToleratedFailures, maxExtraJobsForRescheduling } = settings;
     const failingJobs = await CiJobs.getFailingJobsQueue();
+    this.reservedRetrySlots = 0;
 
     if (failingJobs.length >= 1) {
       const openSpots = await this.determineOpenSpots();
@@ -177,14 +179,23 @@ export class Scheduler {
       }
 
       const ingeminator = new Ingeminator(numberToReschedule, this.gitHub);
-      await ingeminator.rescheduleFailedJobs(failingJobs);
+      const scheduledRetries = await ingeminator.rescheduleFailedJobs(failingJobs);
+      this.reservedRetrySlots = Math.min(scheduledRetries, openSpots);
+
+      const remainingFreshSlots = openSpots - this.reservedRetrySlots;
+      if (remainingFreshSlots > 0) {
+        await Discord.sendDebug(
+          `[Scheduler] Reserved ${this.reservedRetrySlots} slot(s) for retries and kept ${remainingFreshSlots} slot(s) available for fresh jobs.`,
+        );
+      }
     }
 
-    return failingJobs.length <= maxToleratedFailures;
+    const openSpotsAfterRetries = await this.determineOpenSpotsForFreshJobs();
+    return failingJobs.length <= maxToleratedFailures || openSpotsAfterRetries > 0;
   }
 
   async buildLatestEditorImages(): Promise<boolean> {
-    const openSpots = await this.determineOpenSpots();
+    const openSpots = await this.determineOpenSpotsForFreshJobs();
     if (openSpots <= 0) {
       await Discord.sendDebug('[Scheduler] Not scheduling any new jobs, as the queue is full');
       return false;
@@ -246,4 +257,10 @@ export class Scheduler {
     const openSpots = this.maxConcurrentJobs - currentlyRunningJobs;
     return openSpots <= 0 ? 0 : openSpots;
   }
+
+  private async determineOpenSpotsForFreshJobs(): Promise<number> {
+    const openSpots = await this.determineOpenSpots();
+    const availableSpots = openSpots - this.reservedRetrySlots;
+    return availableSpots <= 0 ? 0 : availableSpots;
+  }
 }

functions/src/model/ciBuilds.ts

@@ -116,6 +116,16 @@ export class CiBuilds {
     }));
   };
 
+  public static hasAnyBuildsForJob = async (jobId: string): Promise<boolean> => {
+    const snapshot = await db
+      .collection(CiBuilds.collection)
+      .where('relatedJobId', '==', jobId)
+      .limit(1)
+      .get();
+
+    return snapshot.docs.length > 0;
+  };
+
   /**
    * Registers a new build or handles duplicate dispatches gracefully.
    * Returns the existing status if the build is already in progress or published,

functions/src/model/ciJobs.ts

@@ -126,6 +126,19 @@ export class CiJobs {
     return snapshot.docs.length;
   };
 
+  static getActiveJobs = async (): Promise<CiJobQueue> => {
+    const snapshot = await db
+      .collection(CiJobs.collection)
+      .where('status', 'in', ['scheduled', 'inProgress'])
+      .limit(settings.maxConcurrentJobs)
+      .get();
+
+    return snapshot.docs.map((doc) => ({
+      id: doc.id,
+      data: doc.data() as CiJob,
+    }));
+  };
+
   static create = async (
     jobId: string,
     imageType: ImageType,
@@ -236,6 +249,15 @@ export class CiJobs {
     });
   };
 
+  static resetJobToCreated = async (jobId: string) => {
+    const job = await db.collection(CiJobs.collection).doc(jobId);
+
+    await job.update({
+      status: 'created',
+      modifiedDate: Timestamp.now(),
+    });
+  };
+
   static async removeDryRunJob(jobId: string) {
     if (!jobId.startsWith('dryRun')) {
       throw new Error('Expect only dryRun jobs to be deleted.');