run engine v1: orgs are no longer considered for concurrency

Author: ericallam

apps/webapp/app/env.server.ts

@@ -357,7 +357,7 @@ const EnvironmentSchema = z.object({
   MARQS_AVAILABLE_CAPACITY_BIAS: z.coerce.number().default(0.3),
   MARQS_QUEUE_AGE_RANDOMIZATION_BIAS: z.coerce.number().default(0.25),
   MARQS_REUSE_SNAPSHOT_COUNT: z.coerce.number().int().default(0),
-  MARQS_MAXIMUM_ORG_COUNT: z.coerce.number().int().optional(),
+  MARQS_MAXIMUM_ENV_COUNT: z.coerce.number().int().optional(),
 
   PROD_TASK_HEARTBEAT_INTERVAL_MS: z.coerce.number().int().optional(),
 

apps/webapp/app/routes/projects.v3.$projectRef.metrics/registerProjectMetrics.server.ts

@@ -25,50 +25,6 @@ export async function registerProjectMetrics(
     },
   });
 
-  const firstEnv = allEnvironments[0];
-
-  if (firstEnv) {
-    new Gauge({
-      name: sanitizeMetricName(`trigger_org_queue_concurrency`),
-      help: `The number of tasks currently being executed in the org environment queue`,
-      registers: [registry],
-      async collect() {
-        const length = await marqs?.currentConcurrencyOfOrg(firstEnv);
-
-        if (length) {
-          this.set(length);
-        }
-      },
-    });
-
-    new Gauge({
-      name: sanitizeMetricName(`trigger_org_queue_concurrency_limit`),
-      help: `The concurrency limit for the org queue`,
-      registers: [registry],
-      async collect() {
-        const length = await marqs?.getOrgConcurrencyLimit(firstEnv);
-
-        if (length) {
-          this.set(length);
-        }
-      },
-    });
-
-    new Gauge({
-      name: sanitizeMetricName(`trigger_org_queue_capacity`),
-      help: "The capacity of the org queue",
-      registers: [registry],
-      async collect() {
-        const concurrencyLimit = await marqs?.getOrgConcurrencyLimit(firstEnv);
-        const currentConcurrency = await marqs?.currentConcurrencyOfOrg(firstEnv);
-
-        if (typeof concurrencyLimit === "number" && typeof currentConcurrency === "number") {
-          this.set(concurrencyLimit - currentConcurrency);
-        }
-      },
-    });
-  }
-
   for (const env of allEnvironments) {
     if (env.type === "DEVELOPMENT" && env.orgMember?.userId === userId) {
       await registerEnvironmentMetrics(env, registry);

apps/webapp/app/v3/marqs/concurrencyMonitor.server.ts

@@ -103,18 +103,16 @@ export class MarqsConcurrencyMonitor {
 
   async #processKey(key: string, redis: Redis) {
     key = this.keys.stripKeyPrefix(key);
-    const orgKey = this.keys.orgCurrentConcurrencyKeyFromQueue(key);
     const envKey = this.keys.envCurrentConcurrencyKeyFromQueue(key);
 
     let runIds: string[] = [];
 
     try {
       // Next, we need to get all the items from the key, and any parent keys (org, env, queue) using sunion.
-      runIds = await redis.sunion(orgKey, envKey, key);
+      runIds = await redis.sunion(envKey, key);
     } catch (e) {
       this._logger.error("[MarqsConcurrencyMonitor] error during sunion", {
         key,
-        orgKey,
         envKey,
         runIds,
         error: e,
@@ -136,7 +134,6 @@ export class MarqsConcurrencyMonitor {
     if (completedRunIds.length === 0) {
       this._logger.debug("[MarqsConcurrencyMonitor] no completed runs found", {
         key,
-        orgKey,
         envKey,
         runIds,
         durationMs,
@@ -147,7 +144,6 @@ export class MarqsConcurrencyMonitor {
 
     this._logger.debug("[MarqsConcurrencyMonitor] removing completed runs from queue", {
       key,
-      orgKey,
       envKey,
       completedRunIds,
       durationMs,
@@ -160,15 +156,13 @@ export class MarqsConcurrencyMonitor {
     const pipeline = redis.pipeline();
 
     pipeline.srem(key, ...completedRunIds);
-    pipeline.srem(orgKey, ...completedRunIds);
     pipeline.srem(envKey, ...completedRunIds);
 
     try {
       await pipeline.exec();
     } catch (e) {
       this._logger.error("[MarqsConcurrencyMonitor] error removing completed runs from queue", {
         key,
-        orgKey,
         envKey,
         completedRunIds,
         error: e,

apps/webapp/app/v3/marqs/fairDequeuingStrategy.server.ts

@@ -33,7 +33,6 @@ export type FairDequeuingStrategyBiases = {
 export type FairDequeuingStrategyOptions = {
   redis: Redis;
   keys: MarQSKeyProducer;
-  defaultOrgConcurrency: number;
   defaultEnvConcurrency: number;
   parentQueueLimit: number;
   tracer: Tracer;
@@ -44,7 +43,7 @@ export type FairDequeuingStrategyOptions = {
    */
   biases?: FairDequeuingStrategyBiases;
   reuseSnapshotCount?: number;
-  maximumOrgCount?: number;
+  maximumEnvCount?: number;
 };
 
 type FairQueueConcurrency = {
@@ -56,7 +55,6 @@ type FairQueue = { id: string; age: number; org: string; env: string };
 
 type FairQueueSnapshot = {
   id: string;
-  orgs: Record<string, { concurrency: FairQueueConcurrency }>;
   envs: Record<string, { concurrency: FairQueueConcurrency }>;
   queues: Array<FairQueue>;
 };
@@ -73,7 +71,6 @@ type WeightedQueue = {
 
 const emptyFairQueueSnapshot: FairQueueSnapshot = {
   id: "empty",
-  orgs: {},
   envs: {},
   queues: [],
 };
@@ -124,7 +121,6 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
         const snapshot = await this.#createQueueSnapshot(parentQueue, consumerId);
 
         span.setAttributes({
-          snapshot_org_count: Object.keys(snapshot.orgs).length,
           snapshot_env_count: Object.keys(snapshot.envs).length,
           snapshot_queue_count: snapshot.queues.length,
         });
@@ -344,69 +340,27 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
         return emptyFairQueueSnapshot;
       }
 
-      // Apply org selection if maximumOrgCount is specified
-      let selectedOrgIds: Set<string>;
-      if (this.options.maximumOrgCount && this.options.maximumOrgCount > 0) {
-        selectedOrgIds = this.#selectTopOrgs(queues, this.options.maximumOrgCount);
-        // Filter queues to only include selected orgs
-        queues = queues.filter((queue) => selectedOrgIds.has(queue.org));
+      // Apply env selection if maximumEnvCount is specified
+      let selectedEnvIds: Set<string>;
+      if (this.options.maximumEnvCount && this.options.maximumEnvCount > 0) {
+        selectedEnvIds = this.#selectTopEnvs(queues, this.options.maximumEnvCount);
+        // Filter queues to only include selected envs
+        queues = queues.filter((queue) => selectedEnvIds.has(queue.env));
 
-        span.setAttribute("selected_org_count", selectedOrgIds.size);
+        span.setAttribute("selected_env_count", selectedEnvIds.size);
       }
 
       span.setAttribute("selected_queue_count", queues.length);
 
-      const orgIds = new Set<string>();
       const envIds = new Set<string>();
-      const envIdToOrgId = new Map<string, string>();
 
       for (const queue of queues) {
-        orgIds.add(queue.org);
         envIds.add(queue.env);
-
-        envIdToOrgId.set(queue.env, queue.org);
-      }
-
-      const orgs = await Promise.all(
-        Array.from(orgIds).map(async (orgId) => {
-          return { id: orgId, concurrency: await this.#getOrgConcurrency(orgId) };
-        })
-      );
-
-      const orgsAtFullConcurrency = orgs.filter(
-        (org) => org.concurrency.current >= org.concurrency.limit
-      );
-
-      const orgIdsAtFullConcurrency = new Set(orgsAtFullConcurrency.map((org) => org.id));
-
-      const orgsSnapshot = orgs.reduce((acc, org) => {
-        if (!orgIdsAtFullConcurrency.has(org.id)) {
-          acc[org.id] = org;
-        }
-
-        return acc;
-      }, {} as Record<string, { concurrency: FairQueueConcurrency }>);
-
-      span.setAttributes({
-        org_count: orgs.length,
-        orgs_at_full_concurrency_count: orgsAtFullConcurrency.length,
-        orgs_snapshot_count: Object.keys(orgsSnapshot).length,
-      });
-
-      if (Object.keys(orgsSnapshot).length === 0) {
-        return emptyFairQueueSnapshot;
       }
 
-      const envsWithoutFullOrgs = Array.from(envIds).filter(
-        (envId) => !orgIdsAtFullConcurrency.has(envIdToOrgId.get(envId)!)
-      );
-
       const envs = await Promise.all(
-        envsWithoutFullOrgs.map(async (envId) => {
-          return {
-            id: envId,
-            concurrency: await this.#getEnvConcurrency(envId, envIdToOrgId.get(envId)!),
-          };
+        Array.from(envIds).map(async (envId) => {
+          return { id: envId, concurrency: await this.#getEnvConcurrency(envId) };
         })
       );
 
@@ -420,7 +374,6 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
         if (!envIdsAtFullConcurrency.has(env.id)) {
           acc[env.id] = env;
         }
-
         return acc;
       }, {} as Record<string, { concurrency: FairQueueConcurrency }>);
 
@@ -429,14 +382,10 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
         envs_at_full_concurrency_count: envsAtFullConcurrency.length,
       });
 
-      const queuesSnapshot = queues.filter(
-        (queue) =>
-          !orgIdsAtFullConcurrency.has(queue.org) && !envIdsAtFullConcurrency.has(queue.env)
-      );
+      const queuesSnapshot = queues.filter((queue) => !envIdsAtFullConcurrency.has(queue.env));
 
       const snapshot = {
         id: randomUUID(),
-        orgs: orgsSnapshot,
         envs: envsSnapshot,
         queues: queuesSnapshot,
       };
@@ -455,71 +404,54 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
     });
   }
 
-  #selectTopOrgs(queues: FairQueue[], maximumOrgCount: number): Set<string> {
-    // Group queues by org
-    const queuesByOrg = queues.reduce((acc, queue) => {
-      if (!acc[queue.org]) {
-        acc[queue.org] = [];
+  #selectTopEnvs(queues: FairQueue[], maximumEnvCount: number): Set<string> {
+    // Group queues by env
+    const queuesByEnv = queues.reduce((acc, queue) => {
+      if (!acc[queue.env]) {
+        acc[queue.env] = [];
       }
-      acc[queue.org].push(queue);
+      acc[queue.env].push(queue);
       return acc;
     }, {} as Record<string, FairQueue[]>);
 
-    // Calculate average age for each org
-    const orgAverageAges = Object.entries(queuesByOrg).map(([orgId, orgQueues]) => {
-      const averageAge = orgQueues.reduce((sum, q) => sum + q.age, 0) / orgQueues.length;
-      return { orgId, averageAge };
+    // Calculate average age for each env
+    const envAverageAges = Object.entries(queuesByEnv).map(([envId, envQueues]) => {
+      const averageAge = envQueues.reduce((sum, q) => sum + q.age, 0) / envQueues.length;
+      return { envId, averageAge };
     });
 
     // Perform weighted shuffle based on average ages
-    const maxAge = Math.max(...orgAverageAges.map((o) => o.averageAge));
-    const weightedOrgs = orgAverageAges.map((org) => ({
-      orgId: org.orgId,
-      weight: org.averageAge / maxAge, // Normalize weights
+    const maxAge = Math.max(...envAverageAges.map((e) => e.averageAge));
+    const weightedEnvs = envAverageAges.map((env) => ({
+      envId: env.envId,
+      weight: env.averageAge / maxAge, // Normalize weights
     }));
 
-    // Select top N orgs using weighted shuffle
-    const selectedOrgs = new Set<string>();
-    let remainingOrgs = [...weightedOrgs];
-    let totalWeight = remainingOrgs.reduce((sum, org) => sum + org.weight, 0);
+    // Select top N envs using weighted shuffle
+    const selectedEnvs = new Set<string>();
+    let remainingEnvs = [...weightedEnvs];
+    let totalWeight = remainingEnvs.reduce((sum, env) => sum + env.weight, 0);
 
-    while (selectedOrgs.size < maximumOrgCount && remainingOrgs.length > 0) {
+    while (selectedEnvs.size < maximumEnvCount && remainingEnvs.length > 0) {
       let random = this._rng() * totalWeight;
       let index = 0;
 
-      while (random > 0 && index < remainingOrgs.length) {
-        random -= remainingOrgs[index].weight;
+      while (random > 0 && index < remainingEnvs.length) {
+        random -= remainingEnvs[index].weight;
         index++;
       }
       index = Math.max(0, index - 1);
 
-      selectedOrgs.add(remainingOrgs[index].orgId);
-      totalWeight -= remainingOrgs[index].weight;
-      remainingOrgs.splice(index, 1);
+      selectedEnvs.add(remainingEnvs[index].envId);
+      totalWeight -= remainingEnvs[index].weight;
+      remainingEnvs.splice(index, 1);
     }
 
-    return selectedOrgs;
-  }
-
-  async #getOrgConcurrency(orgId: string): Promise<FairQueueConcurrency> {
-    return await startSpan(this.options.tracer, "getOrgConcurrency", async (span) => {
-      span.setAttribute("org_id", orgId);
-
-      const [currentValue, limitValue] = await Promise.all([
-        this.#getOrgCurrentConcurrency(orgId),
-        this.#getOrgConcurrencyLimit(orgId),
-      ]);
-
-      span.setAttribute("current_value", currentValue);
-      span.setAttribute("limit_value", limitValue);
-
-      return { current: currentValue, limit: limitValue };
-    });
+    return selectedEnvs;
   }
 
-  async #getEnvConcurrency(envId: string, orgId: string): Promise<FairQueueConcurrency> {
+  async #getEnvConcurrency(envId: string): Promise<FairQueueConcurrency> {
     return await startSpan(this.options.tracer, "getEnvConcurrency", async (span) => {
-      span.setAttribute("org_id", orgId);
       span.setAttribute("env_id", envId);
 
       const [currentValue, limitValue] = await Promise.all([
@@ -570,40 +502,6 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
     });
   }
 
-  async #getOrgConcurrencyLimit(orgId: string) {
-    return await startSpan(this.options.tracer, "getOrgConcurrencyLimit", async (span) => {
-      span.setAttribute("org_id", orgId);
-
-      const key = this.options.keys.orgConcurrencyLimitKey(orgId);
-
-      const result = await this._cache.concurrencyLimit.swr(key, async () => {
-        const value = await this.options.redis.get(key);
-
-        if (!value) {
-          return this.options.defaultOrgConcurrency;
-        }
-
-        return Number(value);
-      });
-
-      return result.val ?? this.options.defaultOrgConcurrency;
-    });
-  }
-
-  async #getOrgCurrentConcurrency(orgId: string) {
-    return await startSpan(this.options.tracer, "getOrgCurrentConcurrency", async (span) => {
-      span.setAttribute("org_id", orgId);
-
-      const key = this.options.keys.orgCurrentConcurrencyKey(orgId);
-
-      const result = await this.options.redis.scard(key);
-
-      span.setAttribute("current_value", result);
-
-      return result;
-    });
-  }
-
   async #getEnvConcurrencyLimit(envId: string) {
     return await startSpan(this.options.tracer, "getEnvConcurrencyLimit", async (span) => {
       span.setAttribute("env_id", envId);