fix(run-engine): observe only managed worker groups in the queue-length gauge

Author: ericallam

internal-packages/run-engine/src/engine/index.ts

@@ -485,6 +485,11 @@ export class RunEngine {
    * length even when nothing is dequeuing from it. Includes hidden groups; excludes
    * groups whose cloud provider is configured to be excluded (groups with no cloud
    * provider are always included).
+   *
+   * Only MANAGED groups are observed. UNMANAGED groups are created per project
+   * (masterQueue `<projectId>-<name>`), so observing them would grow the set, and the
+   * per-tick Redis fanout, with the number of self-hosted-worker projects rather than
+   * with the managed regions this gauge is meant to track.
    */
   async refreshWorkerQueueObservation() {
     const suffixes = this.options.workerQueueObserver?.additionalQueueSuffixes ?? [];
@@ -495,6 +500,7 @@ export class RunEngine {
     // Read from the replica: this is a periodic metrics-only read and worker groups change
     // rarely, so a little replication lag is fine and keeps it off the primary.
     const workerGroups = await this.readOnlyPrisma.workerInstanceGroup.findMany({
+      where: { type: "MANAGED" },
       select: { masterQueue: true, cloudProvider: true },
     });
 

internal-packages/run-engine/src/engine/tests/workerQueueObservation.test.ts

@@ -43,6 +43,16 @@ describe("RunEngine worker queue observation", () => {
         },
       });
 
+      // An UNMANAGED (per-project, self-hosted) worker group should not be observed.
+      await prisma.workerInstanceGroup.create({
+        data: {
+          name: "unmanaged-region",
+          masterQueue: "unmanaged-region",
+          type: "UNMANAGED",
+          token: { create: { tokenHash: "unmanaged_region_token_hash" } },
+        },
+      });
+
       const engine = new RunEngine({
         prisma,
         // This test only exercises enqueue + processMasterQueue + the observer gauge, so keep
@@ -118,17 +128,19 @@ describe("RunEngine worker queue observation", () => {
         const defaultBacklog = 3;
         const scheduledBacklog = 2;
         const hiddenBacklog = 2;
-        const doBacklog = 2;
+        const doBacklog = 1;
+        const unmanagedBacklog = 1;
 
         // Build a backlog across several worker queues, then move them into the worker queue
         // lists, but never dequeue.
         await enqueueTo("default", defaultBacklog, "r_default");
         await enqueueTo("default:scheduled", scheduledBacklog, "r_scheduled");
         await enqueueTo("hidden-region", hiddenBacklog, "r_hidden");
         await enqueueTo("do-region", doBacklog, "r_do");
+        await enqueueTo("unmanaged-region", unmanagedBacklog, "r_unmanaged");
         await engine.runQueue.processMasterQueueForEnvironment(
           authenticatedEnvironment.id,
-          defaultBacklog + scheduledBacklog + hiddenBacklog + doBacklog
+          defaultBacklog + scheduledBacklog + hiddenBacklog + doBacklog + unmanagedBacklog
         );
 
         // Observe the worker queues derived from the WorkerInstanceGroup records. No dequeue
@@ -142,6 +154,9 @@ describe("RunEngine worker queue observation", () => {
 
         // Excluded: the DigitalOcean group is not observed even though it has a backlog.
         expect(await lengthOf("do-region")).toBe(0);
+
+        // Excluded: the UNMANAGED (per-project) group is not observed even with a backlog.
+        expect(await lengthOf("unmanaged-region")).toBe(0);
       } finally {
         await engine.quit();
       }