fix: requeue stale active jobs without builds

Author: frostebite

functions/src/logic/buildQueue/cleaner.ts

@@ -1,18 +1,21 @@
 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, 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(latestRepoVersion: string) {
     this.buildsProcessed = 0;
+    await this.requeueActiveJobsWithoutBuilds();
     await this.recoverMaxedOutFailedBuilds(latestRepoVersion);
     await this.reconcileStartedBuildsThatMayHavePublished();
     await this.cleanUpBuildsThatDidntReportBack();
@@ -27,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.

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.');