WIP release concurrency queue

.cursor/mcp.json

@@ -4,4 +4,4 @@
       "url": "http://localhost:3333/sse"
     }
   }
-}
+}

.vscode/launch.json

@@ -138,7 +138,7 @@
       "type": "node-terminal",
       "request": "launch",
       "name": "Debug RunEngine tests",
-      "command": "pnpm run test ./src/engine/tests/delays.test.ts -t 'Delayed run with a ttl'",
+      "command": "pnpm run test ./src/engine/tests/releaseConcurrencyQueue.test.ts -t 'Should manage token bucket and queue correctly'",
       "cwd": "${workspaceFolder}/internal-packages/run-engine",
       "sourceMaps": true
     },

apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.wait.duration.ts

@@ -51,9 +51,7 @@ const { action } = createActionApiRoute(
         environmentId: authentication.environment.id,
         projectId: authentication.environment.project.id,
         organizationId: authentication.environment.organization.id,
-        releaseConcurrency: {
-          releaseQueue: true,
-        },
+        releaseConcurrency: true,
       });
 
       return json({

apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.waitpoints.tokens.$waitpointFriendlyId.wait.ts

@@ -40,6 +40,7 @@ const { action } = createActionApiRoute(
         environmentId: authentication.environment.id,
         projectId: authentication.environment.project.id,
         organizationId: authentication.environment.organization.id,
+        releaseConcurrency: true,
       });
 
       return json<WaitForWaitpointTokenResponseBody>(

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

@@ -63,6 +63,7 @@ import { RunLocker } from "./locking.js";
 import { getMachinePreset } from "./machinePresets.js";
 import { retryOutcomeFromCompletion } from "./retrying.js";
 import {
+  canReleaseConcurrency,
   isCheckpointable,
   isDequeueableExecutionStatus,
   isExecuting,
@@ -70,6 +71,7 @@ import {
   isPendingExecuting,
 } from "./statuses.js";
 import { HeartbeatTimeouts, RunEngineOptions, TriggerParams } from "./types.js";
+import { ReleaseConcurrencyQueue } from "./releaseConcurrencyQueue.js";
 
 const workerCatalog = {
   finishWaitpoint: {
@@ -137,6 +139,7 @@ export class RunEngine {
   private logger = new Logger("RunEngine", "debug");
   private tracer: Tracer;
   private heartbeatTimeouts: HeartbeatTimeouts;
+  private releaseConcurrencyQueue: ReleaseConcurrencyQueue;
   eventBus = new EventEmitter<EventBusEvents>();
 
   constructor(private readonly options: RunEngineOptions) {
@@ -239,6 +242,20 @@ export class RunEngine {
       ...defaultHeartbeatTimeouts,
       ...(options.heartbeatTimeoutsMs ?? {}),
     };
+
+    // Initialize the ReleaseConcurrencyQueue
+    this.releaseConcurrencyQueue = new ReleaseConcurrencyQueue({
+      redis: {
+        ...options.queue.redis, // Use base queue redis options
+        ...options.releaseConcurrency?.redis, // Allow overrides
+        keyPrefix: `${options.queue.redis.keyPrefix}release-concurrency:`,
+      },
+      maxTokens: options.releaseConcurrency?.maxTokens ?? 10, // Default to 10 tokens
+      executor: async (releaseQueue, runId) => {
+        await this.#executeReleasedConcurrencyFromQueue(releaseQueue, runId);
+      },
+      tracer: this.tracer,
+    });
   }
 
   //MARK: - Run functions
@@ -1994,9 +2011,7 @@ export class RunEngine {
     environmentId: string;
     projectId: string;
     organizationId: string;
-    releaseConcurrency?: {
-      releaseQueue: boolean;
-    };
+    releaseConcurrency?: boolean;
     timeout?: Date;
     spanIdToComplete?: string;
     batch?: { id: string; index?: number };
@@ -2096,11 +2111,7 @@ export class RunEngine {
       } else {
         if (releaseConcurrency) {
           //release concurrency
-          await this.runQueue.releaseConcurrency(
-            organizationId,
-            runId,
-            releaseConcurrency.releaseQueue === true
-          );
+          await this.#attemptToReleaseConcurrency(organizationId, snapshot);
         }
       }
 
@@ -2108,12 +2119,76 @@ export class RunEngine {
     });
   }
 
-  // Add releaseConcurrencyIfSuspendedOrGoingToBeSuspended
-  // - Called from blockRunWithWaitpoint when releaseConcurrency exists
-  // - Runlock the run
-  // - Get latest snapshot
-  // - If the run is non suspended or going to be, then bail
-  // - If the run is suspended or going to be, then release the concurrency
+  async #attemptToReleaseConcurrency(orgId: string, snapshot: TaskRunExecutionSnapshot) {
+    // Go ahead and release concurrency immediately if the run is in a development environment
+    if (snapshot.environmentType === "DEVELOPMENT") {
+      return await this.runQueue.releaseConcurrency(orgId, snapshot.runId);
+    }
+
+    const run = await this.prisma.taskRun.findFirst({
+      where: {
+        id: snapshot.runId,
+      },
+      select: {
+        runtimeEnvironment: {
+          select: {
+            id: true,
+            projectId: true,
+            organizationId: true,
+          },
+        },
+      },
+    });
+
+    if (!run) {
+      this.logger.error("Run not found for attemptToReleaseConcurrency", {
+        runId: snapshot.runId,
+      });
+
+      return;
+    }
+
+    await this.releaseConcurrencyQueue.attemptToRelease(
+      this.runQueue.keys.releaseConcurrencyKey({
+        orgId: run.runtimeEnvironment.organizationId,
+        projectId: run.runtimeEnvironment.projectId,
+        envId: run.runtimeEnvironment.id,
+      }),
+      snapshot.runId
+    );
+
+    return;
+  }
+
+  async #executeReleasedConcurrencyFromQueue(releaseQueue: string, runId: string) {
+    const releaseQueueDescriptor =
+      this.runQueue.keys.releaseConcurrencyDescriptorFromQueue(releaseQueue);
+
+    this.logger.debug("Executing released concurrency", {
+      releaseQueue,
+      runId,
+      releaseQueueDescriptor,
+    });
+
+    // - Runlock the run
+    // - Get latest snapshot
+    // - If the run is non suspended or going to be, then bail
+    // - If the run is suspended or going to be, then release the concurrency
+    await this.runLock.lock([runId], 5_000, async (signal) => {
+      const snapshot = await getLatestExecutionSnapshot(this.prisma, runId);
+
+      if (!canReleaseConcurrency(snapshot.executionStatus)) {
+        this.logger.debug("Run is not in a state to release concurrency", {
+          runId,
+          snapshot,
+        });
+
+        return;
+      }
+
+      return await this.runQueue.releaseConcurrency(releaseQueueDescriptor.orgId, snapshot.runId);
+    });
+  }
 
   /** This completes a waitpoint and updates all entries so the run isn't blocked,
    * if they're no longer blocked. This doesn't suffer from race conditions. */
@@ -2300,6 +2375,7 @@ export class RunEngine {
             select: {
               id: true,
               projectId: true,
+              organizationId: true,
             },
           },
         },
@@ -2340,6 +2416,16 @@ export class RunEngine {
         runnerId,
       });
 
+      // Refill the token bucket for the release concurrency queue
+      await this.releaseConcurrencyQueue.refillTokens(
+        this.runQueue.keys.releaseConcurrencyKey({
+          orgId: run.runtimeEnvironment.organizationId,
+          projectId: run.runtimeEnvironment.projectId,
+          envId: run.runtimeEnvironment.id,
+        }),
+        1
+      );
+
       return {
         ok: true as const,
         ...executionResultFromSnapshot(newSnapshot),

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

@@ -0,0 +1,533 @@
+import { Callback, createRedisClient, Redis, Result, type RedisOptions } from "@internal/redis";
+import { Tracer } from "@internal/tracing";
+import { Logger } from "@trigger.dev/core/logger";
+import { setTimeout } from "node:timers/promises";
+import { z } from "zod";
+
+export type ReleaseConcurrencyQueueRetryOptions = {
+  maxRetries?: number;
+  backoff?: {
+    minDelay?: number; // Defaults to 1000
+    maxDelay?: number; // Defaults to 60000
+    factor?: number; // Defaults to 2
+  };
+};
+
+export type ReleaseConcurrencyQueueOptions<T> = {
+  redis: RedisOptions;
+  executor: (releaseQueue: T, runId: string) => Promise<void>;
+  keys: {
+    fromDescriptor: (releaseQueue: T) => string;
+    toDescriptor: (releaseQueue: string) => T;
+  };
+  consumersCount?: number;
+  masterQueuesKey?: string;
+  tracer?: Tracer;
+  logger?: Logger;
+  pollInterval?: number;
+  batchSize?: number;
+  retry?: ReleaseConcurrencyQueueRetryOptions;
+};
+
+const QueueItemMetadata = z.object({
+  retryCount: z.number(),
+  lastAttempt: z.number(),
+});
+
+type QueueItemMetadata = z.infer<typeof QueueItemMetadata>;
+
+export class ReleaseConcurrencyQueue<T> {
+  private redis: Redis;
+  private logger: Logger;
+
+  private keyPrefix: string;
+  private masterQueuesKey: string;
+  private consumersCount: number;
+  private pollInterval: number;
+  private keys: ReleaseConcurrencyQueueOptions<T>["keys"];
+  private consumersEnabled: boolean;
+  private batchSize: number;
+  private maxRetries: number;
+  private backoff: NonNullable<Required<ReleaseConcurrencyQueueRetryOptions["backoff"]>>;
+
+  constructor(private readonly options: ReleaseConcurrencyQueueOptions<T>) {
+    this.redis = createRedisClient(options.redis);
+    this.keyPrefix = options.redis.keyPrefix ?? "re2:release-concurrency-queue:";
+    this.logger = options.logger ?? new Logger("ReleaseConcurrencyQueue");
+
+    this.masterQueuesKey = options.masterQueuesKey ?? "master-queue";
+    this.consumersCount = options.consumersCount ?? 1;
+    this.pollInterval = options.pollInterval ?? 1000;
+    this.keys = options.keys;
+    this.batchSize = options.batchSize ?? 5;
+    this.maxRetries = options.retry?.maxRetries ?? 3;
+    this.backoff = {
+      minDelay: options.retry?.backoff?.minDelay ?? 1000,
+      maxDelay: options.retry?.backoff?.maxDelay ?? 60000,
+      factor: options.retry?.backoff?.factor ?? 2,
+    };
+
+    this.consumersEnabled = true;
+
+    this.#registerCommands();
+    this.#startConsumers();
+  }
+
+  public async quit() {
+    this.consumersEnabled = false;
+    await this.redis.quit();
+  }
+
+  /**
+   * Attempt to release concurrency for a run.
+   *
+   * If there is a token available, then immediately release the concurrency
+   * If there is no token available, then we'll add the operation to a queue
+   * and wait until the token is available.
+   */
+  public async attemptToRelease(releaseQueueDescriptor: T, runId: string, maxTokens: number) {
+    const releaseQueue = this.keys.fromDescriptor(releaseQueueDescriptor);
+
+    const result = await this.redis.consumeToken(
+      this.masterQueuesKey,
+      this.#bucketKey(releaseQueue),
+      this.#queueKey(releaseQueue),
+      this.#metadataKey(releaseQueue),
+      releaseQueue,
+      runId,
+      String(maxTokens),
+      String(Date.now())
+    );
+
+    if (!!result) {
+      await this.#callExecutor(releaseQueueDescriptor, runId, {
+        retryCount: 0,
+        lastAttempt: Date.now(),
+      });
+    }
+  }
+
+  /**
+   * Refill the token bucket for a release queue.
+   *
+   * This will add the amount of tokens to the token bucket.
+   */
+  public async refillTokens(releaseQueueDescriptor: T, maxTokens: number, amount: number = 1) {
+    const releaseQueue = this.keys.fromDescriptor(releaseQueueDescriptor);
+
+    if (amount < 0) {
+      throw new Error("Cannot refill with negative tokens");
+    }
+
+    if (amount === 0) {
+      return [];
+    }
+
+    await this.redis.refillTokens(
+      this.masterQueuesKey,
+      this.#bucketKey(releaseQueue),
+      this.#queueKey(releaseQueue),
+      releaseQueue,
+      String(amount),
+      String(maxTokens)
+    );
+  }
+
+  /**
+   * Get the next queue that has available capacity and process one item from it
+   * Returns true if an item was processed, false if no items were available
+   */
+  public async processNextAvailableQueue(): Promise<boolean> {
+    const result = await this.redis.processMasterQueue(
+      this.masterQueuesKey,
+      this.keyPrefix,
+      this.batchSize,
+      String(Date.now())
+    );
+
+    if (!result || result.length === 0) {
+      return false;
+    }
+
+    await Promise.all(
+      result.map(([queue, runId, metadata]) => {
+        const itemMetadata = QueueItemMetadata.parse(JSON.parse(metadata));
+        const releaseQueueDescriptor = this.keys.toDescriptor(queue);
+        return this.#callExecutor(releaseQueueDescriptor, runId, itemMetadata);
+      })
+    );
+
+    return true;
+  }
+
+  async #callExecutor(releaseQueueDescriptor: T, runId: string, metadata: QueueItemMetadata) {
+    try {
+      this.logger.info("Executing run:", { releaseQueueDescriptor, runId });
+
+      await this.options.executor(releaseQueueDescriptor, runId);
+    } catch (error) {
+      this.logger.error("Error executing run:", { error });
+
+      if (metadata.retryCount >= this.maxRetries) {
+        this.logger.error("Max retries reached:", {
+          releaseQueueDescriptor,
+          runId,
+          retryCount: metadata.retryCount,
+        });
+
+        // Return the token but don't requeue
+        const releaseQueue = this.keys.fromDescriptor(releaseQueueDescriptor);
+        await this.redis.returnTokenOnly(
+          this.masterQueuesKey,
+          this.#bucketKey(releaseQueue),
+          this.#queueKey(releaseQueue),
+          this.#metadataKey(releaseQueue),
+          releaseQueue,
+          runId
+        );
+
+        this.logger.info("Returned token:", { releaseQueueDescriptor, runId });
+
+        return;
+      }
+
+      const updatedMetadata: QueueItemMetadata = {
+        ...metadata,
+        retryCount: metadata.retryCount + 1,
+        lastAttempt: Date.now(),
+      };
+
+      const releaseQueue = this.keys.fromDescriptor(releaseQueueDescriptor);
+
+      await this.redis.returnTokenAndRequeue(
+        this.masterQueuesKey,
+        this.#bucketKey(releaseQueue),
+        this.#queueKey(releaseQueue),
+        this.#metadataKey(releaseQueue),
+        releaseQueue,
+        runId,
+        JSON.stringify(updatedMetadata),
+        this.#calculateBackoffScore(updatedMetadata)
+      );
+    }
+  }
+
+  #bucketKey(releaseQueue: string) {
+    return `${releaseQueue}:bucket`;
+  }
+
+  #queueKey(releaseQueue: string) {
+    return `${releaseQueue}:queue`;
+  }
+
+  #metadataKey(releaseQueue: string) {
+    return `${releaseQueue}:metadata`;
+  }
+
+  #startConsumers() {
+    for (let i = 0; i < this.consumersCount; i++) {
+      this.#startConsumer();
+    }
+  }
+
+  async #startConsumer() {
+    while (this.consumersEnabled) {
+      try {
+        const processed = await this.processNextAvailableQueue();
+        if (!processed) {
+          // No items available, wait before trying again
+          await setTimeout(this.pollInterval);
+        }
+      } catch (error) {
+        // Handle error, maybe wait before retrying
+        this.logger.error("Error processing queue:", { error });
+        await setTimeout(this.pollInterval);
+      }
+    }
+  }
+
+  #calculateBackoffScore(item: QueueItemMetadata): string {
+    const delay = Math.min(
+      this.backoff.maxDelay,
+      this.backoff.minDelay * Math.pow(this.backoff.factor, item.retryCount)
+    );
+    return String(Date.now() + delay);
+  }
+
+  #registerCommands() {
+    this.redis.defineCommand("consumeToken", {
+      numberOfKeys: 4,
+      lua: `
+local masterQueuesKey = KEYS[1]
+local bucketKey = KEYS[2]
+local queueKey = KEYS[3]
+local metadataKey = KEYS[4]
+
+local releaseQueue = ARGV[1]
+local runId = ARGV[2]
+local maxTokens = tonumber(ARGV[3])
+local score = ARGV[4]
+
+-- Get the current token count
+local currentTokens = tonumber(redis.call("GET", bucketKey) or maxTokens)
+
+-- If we have enough tokens, then consume them
+if currentTokens >= 1 then
+  redis.call("SET", bucketKey, currentTokens - 1)
+  redis.call("ZREM", queueKey, runId)
+
+  -- Clean up metadata when successfully consuming
+  redis.call("HDEL", metadataKey, runId)
+
+  -- Get queue length after removing the item
+  local queueLength = redis.call("ZCARD", queueKey)
+
+  -- If we still have tokens and items in queue, update available queues
+  if currentTokens > 0 and queueLength > 0 then
+    redis.call("ZADD", masterQueuesKey, currentTokens, releaseQueue)
+  else
+    redis.call("ZREM", masterQueuesKey, releaseQueue)
+  end
+
+  return true
+end
+
+-- If we don't have enough tokens, then we need to add the operation to the queue
+redis.call("ZADD", queueKey, score, runId)
+
+-- Initialize or update metadata
+local metadata = cjson.encode({
+  retryCount = 0,
+  lastAttempt = tonumber(score)
+})
+redis.call("HSET", metadataKey, runId, metadata)
+
+-- Remove from the master queue
+redis.call("ZREM", masterQueuesKey, releaseQueue)
+
+return false
+      `,
+    });
+
+    this.redis.defineCommand("refillTokens", {
+      numberOfKeys: 3,
+      lua: `
+local masterQueuesKey = KEYS[1]
+local bucketKey = KEYS[2]
+local queueKey = KEYS[3]
+
+local releaseQueue = ARGV[1]
+local amount = tonumber(ARGV[2])
+local maxTokens = tonumber(ARGV[3])
+
+local currentTokens = tonumber(redis.call("GET", bucketKey) or maxTokens)
+
+-- Add the amount of tokens to the token bucket
+local newTokens = currentTokens + amount
+
+-- If we have more tokens than the max, then set the token bucket to the max
+if newTokens > maxTokens then
+  newTokens = maxTokens
+end
+
+redis.call("SET", bucketKey, newTokens)
+
+-- Get the number of items in the queue
+local queueLength = redis.call("ZCARD", queueKey)
+
+-- If we have tokens available and items in the queue, add to available queues
+if newTokens > 0 and queueLength > 0 then
+  redis.call("ZADD", masterQueuesKey, newTokens, releaseQueue)
+else
+  redis.call("ZREM", masterQueuesKey, releaseQueue)
+end
+      `,
+    });
+
+    this.redis.defineCommand("processMasterQueue", {
+      numberOfKeys: 1,
+      lua: `
+local masterQueuesKey = KEYS[1]
+
+local keyPrefix = ARGV[1]
+local batchSize = tonumber(ARGV[2])
+local currentTime = tonumber(ARGV[3])
+-- Get the queue with the highest number of available tokens
+local queues = redis.call("ZREVRANGE", masterQueuesKey, 0, 0, "WITHSCORES")
+if #queues == 0 then
+  return nil
+end
+
+local queueName = queues[1]
+local availableTokens = tonumber(queues[2])
+
+local bucketKey = keyPrefix .. queueName .. ":bucket"
+local queueKey = keyPrefix .. queueName .. ":queue"
+local metadataKey = keyPrefix .. queueName .. ":metadata"
+
+-- Get the oldest item from the queue
+local items = redis.call("ZRANGEBYSCORE", queueKey, 0, currentTime, "LIMIT", 0, batchSize - 1)
+if #items == 0 then
+-- No items ready to be processed yet
+  return nil
+end
+
+-- Calculate how many items we can actually process
+local itemsToProcess = math.min(#items, availableTokens)
+local results = {}
+
+-- Consume tokens and collect results
+local currentTokens = tonumber(redis.call("GET", bucketKey))
+redis.call("SET", bucketKey, currentTokens - itemsToProcess)
+
+-- Remove the items from the queue and add to results
+for i = 1, itemsToProcess do
+  local runId = items[i]
+  redis.call("ZREM", queueKey, runId)
+
+  -- Get metadata before removing it
+  local metadata = redis.call("HGET", metadataKey, runId)
+  redis.call("HDEL", metadataKey, runId)
+
+  table.insert(results, { queueName, runId, metadata })
+end
+
+-- Get remaining queue length
+local queueLength = redis.call("ZCARD", queueKey)
+
+-- Update available queues score or remove if no more tokens
+local remainingTokens = currentTokens - itemsToProcess
+if remainingTokens > 0 and queueLength > 0 then
+  redis.call("ZADD", masterQueuesKey, remainingTokens, queueName)
+else
+  redis.call("ZREM", masterQueuesKey, queueName)
+end
+
+return results
+      `,
+    });
+
+    this.redis.defineCommand("returnTokenAndRequeue", {
+      numberOfKeys: 4,
+      lua: `
+local masterQueuesKey = KEYS[1]
+local bucketKey = KEYS[2]
+local queueKey = KEYS[3]  
+local metadataKey = KEYS[4]
+
+local releaseQueue = ARGV[1]
+local runId = ARGV[2]
+local metadata = ARGV[3]
+local score = ARGV[4]
+
+--  Return the token to the bucket
+local currentTokens = tonumber(redis.call("GET", bucketKey))
+local remainingTokens = currentTokens + 1
+redis.call("SET", bucketKey, remainingTokens)
+
+-- Add the item back to the queue
+redis.call("ZADD", queueKey, score, runId)
+
+-- Add the metadata back to the item
+redis.call("HSET", metadataKey, runId, metadata)
+
+-- Update the master queue
+local queueLength = redis.call("ZCARD", queueKey)
+if queueLength > 0 then
+  redis.call("ZADD", masterQueuesKey, remainingTokens, releaseQueue)
+else
+  redis.call("ZREM", masterQueuesKey, releaseQueue)
+end
+
+return true
+      `,
+    });
+
+    this.redis.defineCommand("returnTokenOnly", {
+      numberOfKeys: 4,
+      lua: `
+local masterQueuesKey = KEYS[1]
+local bucketKey = KEYS[2]
+local queueKey = KEYS[3]
+local metadataKey = KEYS[4]
+
+local releaseQueue = ARGV[1]
+local runId = ARGV[2]
+
+-- Return the token to the bucket
+local currentTokens = tonumber(redis.call("GET", bucketKey))
+local remainingTokens = currentTokens + 1
+redis.call("SET", bucketKey, remainingTokens)
+
+-- Clean up metadata
+redis.call("HDEL", metadataKey, runId)
+
+-- Update the master queue based on remaining queue length
+local queueLength = redis.call("ZCARD", queueKey)
+if queueLength > 0 then
+  redis.call("ZADD", masterQueuesKey, remainingTokens, releaseQueue)
+else
+  redis.call("ZREM", masterQueuesKey, releaseQueue)
+end
+
+return true
+      `,
+    });
+  }
+}
+
+declare module "@internal/redis" {
+  interface RedisCommander<Context> {
+    consumeToken(
+      masterQueuesKey: string,
+      bucketKey: string,
+      queueKey: string,
+      metadataKey: string,
+      releaseQueue: string,
+      runId: string,
+      maxTokens: string,
+      score: string,
+      callback?: Callback<string>
+    ): Result<string, Context>;
+
+    refillTokens(
+      masterQueuesKey: string,
+      bucketKey: string,
+      queueKey: string,
+      releaseQueue: string,
+      amount: string,
+      maxTokens: string,
+      callback?: Callback<void>
+    ): Result<void, Context>;
+
+    processMasterQueue(
+      masterQueuesKey: string,
+      keyPrefix: string,
+      batchSize: number,
+      currentTime: string,
+      callback?: Callback<[string, string, string][]>
+    ): Result<[string, string, string][], Context>;
+
+    returnTokenAndRequeue(
+      masterQueuesKey: string,
+      bucketKey: string,
+      queueKey: string,
+      metadataKey: string,
+      releaseQueue: string,
+      runId: string,
+      metadata: string,
+      score: string,
+      callback?: Callback<void>
+    ): Result<void, Context>;
+
+    returnTokenOnly(
+      masterQueuesKey: string,
+      bucketKey: string,
+      queueKey: string,
+      metadataKey: string,
+      releaseQueue: string,
+      runId: string,
+      callback?: Callback<void>
+    ): Result<void, Context>;
+  }
+}

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

@@ -44,3 +44,8 @@ export function isFinalRunStatus(status: TaskRunStatus): boolean {
 
   return finalStatuses.includes(status);
 }
+
+export function canReleaseConcurrency(status: TaskRunExecutionStatus): boolean {
+  const releaseableStatuses: TaskRunExecutionStatus[] = ["SUSPENDED", "EXECUTING_WITH_WAITPOINTS"];
+  return releaseableStatuses.includes(status);
+}

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

@@ -0,0 +1,544 @@
+import { redisTest, StartedRedisContainer } from "@internal/testcontainers";
+import { ReleaseConcurrencyQueue } from "../releaseConcurrencyQueue.js";
+import { setTimeout } from "node:timers/promises";
+
+type TestQueueDescriptor = {
+  name: string;
+};
+
+function createReleaseConcurrencyQueue(redisContainer: StartedRedisContainer) {
+  const executedRuns: { releaseQueue: string; runId: string }[] = [];
+
+  const queue = new ReleaseConcurrencyQueue<TestQueueDescriptor>({
+    redis: {
+      keyPrefix: "release-queue:test:",
+      host: redisContainer.getHost(),
+      port: redisContainer.getPort(),
+    },
+    executor: async (releaseQueue, runId) => {
+      executedRuns.push({ releaseQueue: releaseQueue.name, runId });
+    },
+    keys: {
+      fromDescriptor: (descriptor) => descriptor.name,
+      toDescriptor: (name) => ({ name }),
+    },
+    pollInterval: 100,
+  });
+
+  return {
+    queue,
+    executedRuns,
+  };
+}
+
+describe("ReleaseConcurrencyQueue", () => {
+  redisTest("Should manage token bucket and queue correctly", async ({ redisContainer }) => {
+    const { queue, executedRuns } = createReleaseConcurrencyQueue(redisContainer);
+
+    try {
+      // First two attempts should execute immediately (we have 2 tokens)
+      await queue.attemptToRelease({ name: "test-queue" }, "run1", 2);
+      await queue.attemptToRelease({ name: "test-queue" }, "run2", 2);
+
+      // Verify first two runs were executed
+      expect(executedRuns).toHaveLength(2);
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run1" });
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run2" });
+
+      // Third attempt should be queued (no tokens left)
+      await queue.attemptToRelease({ name: "test-queue" }, "run3", 2);
+      expect(executedRuns).toHaveLength(2); // Still 2, run3 is queued
+
+      // Refill one token, should execute run3
+      await queue.refillTokens({ name: "test-queue" }, 2, 1);
+
+      // Now we need to wait for the queue to be processed
+      await setTimeout(1000);
+
+      expect(executedRuns).toHaveLength(3);
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run3" });
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should handle multiple refills correctly", async ({ redisContainer }) => {
+    const { queue, executedRuns } = createReleaseConcurrencyQueue(redisContainer);
+
+    try {
+      // Queue up 5 runs (more than maxTokens)
+      await queue.attemptToRelease({ name: "test-queue" }, "run1", 3);
+      await queue.attemptToRelease({ name: "test-queue" }, "run2", 3);
+      await queue.attemptToRelease({ name: "test-queue" }, "run3", 3);
+      await queue.attemptToRelease({ name: "test-queue" }, "run4", 3);
+      await queue.attemptToRelease({ name: "test-queue" }, "run5", 3);
+
+      // First 3 should be executed immediately (maxTokens = 3)
+      expect(executedRuns).toHaveLength(3);
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run1" });
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run2" });
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run3" });
+
+      // Refill 2 tokens
+      await queue.refillTokens({ name: "test-queue" }, 3, 2);
+
+      await setTimeout(1000);
+
+      // Should execute the remaining 2 runs
+      expect(executedRuns).toHaveLength(5);
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run4" });
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run5" });
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should handle multiple queues independently", async ({ redisContainer }) => {
+    const { queue, executedRuns } = createReleaseConcurrencyQueue(redisContainer);
+
+    try {
+      // Add runs to different queues
+      await queue.attemptToRelease({ name: "queue1" }, "run1", 1);
+      await queue.attemptToRelease({ name: "queue1" }, "run2", 1);
+      await queue.attemptToRelease({ name: "queue2" }, "run3", 1);
+      await queue.attemptToRelease({ name: "queue2" }, "run4", 1);
+
+      // Only first run from each queue should be executed
+      expect(executedRuns).toHaveLength(2);
+      expect(executedRuns).toContainEqual({ releaseQueue: "queue1", runId: "run1" });
+      expect(executedRuns).toContainEqual({ releaseQueue: "queue2", runId: "run3" });
+
+      // Refill tokens for queue1
+      await queue.refillTokens({ name: "queue1" }, 1, 1);
+
+      await setTimeout(1000);
+
+      // Should only execute the queued run from queue1
+      expect(executedRuns).toHaveLength(3);
+      expect(executedRuns).toContainEqual({ releaseQueue: "queue1", runId: "run2" });
+
+      // Refill tokens for queue2
+      await queue.refillTokens({ name: "queue2" }, 1, 1);
+
+      await setTimeout(1000);
+
+      // Should execute the queued run from queue2
+      expect(executedRuns).toHaveLength(4);
+      expect(executedRuns).toContainEqual({ releaseQueue: "queue2", runId: "run4" });
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should not allow refilling more than maxTokens", async ({ redisContainer }) => {
+    const { queue, executedRuns } = createReleaseConcurrencyQueue(redisContainer);
+
+    try {
+      // Add two runs
+      await queue.attemptToRelease({ name: "test-queue" }, "run1", 1);
+      await queue.attemptToRelease({ name: "test-queue" }, "run2", 1);
+
+      // First run should be executed immediately
+      expect(executedRuns).toHaveLength(1);
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run1" });
+
+      // Refill with more tokens than needed
+      await queue.refillTokens({ name: "test-queue" }, 1, 5);
+
+      await setTimeout(1000);
+
+      // Should only execute the one remaining run
+      expect(executedRuns).toHaveLength(2);
+      expect(executedRuns).toContainEqual({ releaseQueue: "test-queue", runId: "run2" });
+
+      // Add another run - should NOT execute immediately because we don't have excess tokens
+      await queue.attemptToRelease({ name: "test-queue" }, "run3", 1);
+      expect(executedRuns).toHaveLength(2);
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should maintain FIFO order when releasing", async ({ redisContainer }) => {
+    const { queue, executedRuns } = createReleaseConcurrencyQueue(redisContainer);
+
+    try {
+      // Queue up multiple runs
+      await queue.attemptToRelease({ name: "test-queue" }, "run1", 1);
+      await queue.attemptToRelease({ name: "test-queue" }, "run2", 1);
+      await queue.attemptToRelease({ name: "test-queue" }, "run3", 1);
+      await queue.attemptToRelease({ name: "test-queue" }, "run4", 1);
+
+      // First run should be executed immediately
+      expect(executedRuns).toHaveLength(1);
+      expect(executedRuns[0]).toEqual({ releaseQueue: "test-queue", runId: "run1" });
+
+      // Refill tokens one at a time and verify order
+      await queue.refillTokens({ name: "test-queue" }, 1, 1);
+
+      await setTimeout(1000);
+
+      expect(executedRuns).toHaveLength(2);
+      expect(executedRuns[1]).toEqual({ releaseQueue: "test-queue", runId: "run2" });
+
+      await queue.refillTokens({ name: "test-queue" }, 1, 1);
+
+      await setTimeout(1000);
+
+      expect(executedRuns).toHaveLength(3);
+      expect(executedRuns[2]).toEqual({ releaseQueue: "test-queue", runId: "run3" });
+
+      await queue.refillTokens({ name: "test-queue" }, 1, 1);
+
+      await setTimeout(1000);
+
+      expect(executedRuns).toHaveLength(4);
+      expect(executedRuns[3]).toEqual({ releaseQueue: "test-queue", runId: "run4" });
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest(
+    "Should handle executor failures by returning the token and adding the item into the queue",
+    async ({ redisContainer }) => {
+      let shouldFail = true;
+
+      const executedRuns: { releaseQueue: string; runId: string }[] = [];
+
+      const queue = new ReleaseConcurrencyQueue<string>({
+        redis: {
+          keyPrefix: "release-queue:test:",
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+        },
+        executor: async (releaseQueue, runId) => {
+          if (shouldFail) {
+            throw new Error("Executor failed");
+          }
+          executedRuns.push({ releaseQueue, runId });
+        },
+        keys: {
+          fromDescriptor: (descriptor) => descriptor,
+          toDescriptor: (name) => name,
+        },
+        batchSize: 2,
+      });
+
+      try {
+        // Attempt to release with failing executor
+        await queue.attemptToRelease("test-queue", "run1", 2);
+        // Does not execute because the executor throws an error
+        expect(executedRuns).toHaveLength(0);
+
+        // Token should have been returned to the bucket so this should try to execute immediately and fail again
+        await queue.attemptToRelease("test-queue", "run2", 2);
+        expect(executedRuns).toHaveLength(0);
+
+        // Allow executor to succeed
+        shouldFail = false;
+
+        await setTimeout(1000);
+
+        // Should now execute successfully
+        expect(executedRuns).toHaveLength(2);
+        expect(executedRuns[0]).toEqual({ releaseQueue: "test-queue", runId: "run1" });
+        expect(executedRuns[1]).toEqual({ releaseQueue: "test-queue", runId: "run2" });
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
+
+  redisTest("Should handle invalid token amounts", async ({ redisContainer }) => {
+    const { queue, executedRuns } = createReleaseConcurrencyQueue(redisContainer);
+
+    try {
+      // Try to refill with negative tokens
+      await expect(queue.refillTokens({ name: "test-queue" }, 1, -1)).rejects.toThrow();
+
+      // Try to refill with zero tokens
+      await queue.refillTokens({ name: "test-queue" }, 1, 0);
+
+      await setTimeout(1000);
+
+      expect(executedRuns).toHaveLength(0);
+
+      // Verify normal operation still works
+      await queue.attemptToRelease({ name: "test-queue" }, "run1", 1);
+      expect(executedRuns).toHaveLength(1);
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should handle concurrent operations correctly", async ({ redisContainer }) => {
+    const executedRuns: { releaseQueue: string; runId: string }[] = [];
+
+    const queue = new ReleaseConcurrencyQueue<string>({
+      redis: {
+        keyPrefix: "release-queue:test:",
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+      },
+      executor: async (releaseQueue, runId) => {
+        // Add small delay to simulate work
+        await setTimeout(10);
+        executedRuns.push({ releaseQueue, runId });
+      },
+      keys: {
+        fromDescriptor: (descriptor) => descriptor,
+        toDescriptor: (name) => name,
+      },
+      batchSize: 5,
+    });
+
+    try {
+      // Attempt multiple concurrent releases
+      await Promise.all([
+        queue.attemptToRelease("test-queue", "run1", 2),
+        queue.attemptToRelease("test-queue", "run2", 2),
+        queue.attemptToRelease("test-queue", "run3", 2),
+        queue.attemptToRelease("test-queue", "run4", 2),
+      ]);
+
+      // Should only execute maxTokens (2) runs
+      expect(executedRuns).toHaveLength(2);
+
+      // Attempt concurrent refills
+      queue.refillTokens("test-queue", 2, 2);
+
+      await setTimeout(1000);
+
+      // Should execute remaining runs
+      expect(executedRuns).toHaveLength(4);
+
+      // Verify all runs were executed exactly once
+      const runCounts = executedRuns.reduce(
+        (acc, { runId }) => {
+          acc[runId] = (acc[runId] || 0) + 1;
+          return acc;
+        },
+        {} as Record<string, number>
+      );
+
+      Object.values(runCounts).forEach((count) => {
+        expect(count).toBe(1);
+      });
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should clean up Redis resources on quit", async ({ redisContainer }) => {
+    const { queue } = createReleaseConcurrencyQueue(redisContainer);
+
+    // Add some data
+    await queue.attemptToRelease({ name: "test-queue" }, "run1", 1);
+    await queue.attemptToRelease({ name: "test-queue" }, "run2", 1);
+
+    // Quit the queue
+    await queue.quit();
+
+    // Verify we can't perform operations after quit
+    await expect(queue.attemptToRelease({ name: "test-queue" }, "run3", 1)).rejects.toThrow();
+    await expect(queue.refillTokens({ name: "test-queue" }, 1, 1)).rejects.toThrow();
+  });
+
+  redisTest("Should stop retrying after max retries is reached", async ({ redisContainer }) => {
+    let failCount = 0;
+    const executedRuns: { releaseQueue: string; runId: string; attempt: number }[] = [];
+
+    const queue = new ReleaseConcurrencyQueue<string>({
+      redis: {
+        keyPrefix: "release-queue:test:",
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+      },
+      executor: async (releaseQueue, runId) => {
+        failCount++;
+        executedRuns.push({ releaseQueue, runId, attempt: failCount });
+        throw new Error("Executor failed");
+      },
+      keys: {
+        fromDescriptor: (descriptor) => descriptor,
+        toDescriptor: (name) => name,
+      },
+      retry: {
+        maxRetries: 2, // Set max retries to 2 (will attempt 3 times total: initial + 2 retries)
+      },
+      pollInterval: 100, // Reduce poll interval for faster test
+    });
+
+    try {
+      // Attempt to release - this will fail and retry
+      await queue.attemptToRelease("test-queue", "run1", 1);
+
+      // Wait for retries to occur
+      await setTimeout(2000);
+
+      // Should have attempted exactly 3 times (initial + 2 retries)
+      expect(executedRuns).toHaveLength(3);
+      expect(executedRuns[0]).toEqual({ releaseQueue: "test-queue", runId: "run1", attempt: 1 });
+      expect(executedRuns[1]).toEqual({ releaseQueue: "test-queue", runId: "run1", attempt: 2 });
+      expect(executedRuns[2]).toEqual({ releaseQueue: "test-queue", runId: "run1", attempt: 3 });
+
+      // Verify that no more retries occur
+      await setTimeout(1000);
+      expect(executedRuns).toHaveLength(3); // Should still be 3
+
+      // Attempt a new release to verify the token was returned
+      let secondRunAttempted = false;
+      const queue2 = new ReleaseConcurrencyQueue<string>({
+        redis: {
+          keyPrefix: "release-queue:test:",
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+        },
+        executor: async (releaseQueue, runId) => {
+          secondRunAttempted = true;
+        },
+        keys: {
+          fromDescriptor: (descriptor) => descriptor,
+          toDescriptor: (name) => name,
+        },
+      });
+
+      await queue2.attemptToRelease("test-queue", "run2", 1);
+      expect(secondRunAttempted).toBe(true); // Should execute immediately because token was returned
+
+      await queue2.quit();
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should handle max retries in batch processing", async ({ redisContainer }) => {
+    const executedRuns: { releaseQueue: string; runId: string; attempt: number }[] = [];
+    const runAttempts: Record<string, number> = {};
+
+    const queue = new ReleaseConcurrencyQueue<string>({
+      redis: {
+        keyPrefix: "release-queue:test:",
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+      },
+      executor: async (releaseQueue, runId) => {
+        runAttempts[runId] = (runAttempts[runId] || 0) + 1;
+        executedRuns.push({ releaseQueue, runId, attempt: runAttempts[runId] });
+        throw new Error("Executor failed");
+      },
+      keys: {
+        fromDescriptor: (descriptor) => descriptor,
+        toDescriptor: (name) => name,
+      },
+      retry: {
+        maxRetries: 2,
+      },
+      batchSize: 3,
+      pollInterval: 100,
+    });
+
+    try {
+      // Queue up multiple runs
+      await Promise.all([
+        queue.attemptToRelease("test-queue", "run1", 3),
+        queue.attemptToRelease("test-queue", "run2", 3),
+        queue.attemptToRelease("test-queue", "run3", 3),
+      ]);
+
+      // Wait for all retries to complete
+      await setTimeout(2000);
+
+      // Each run should have been attempted exactly 3 times
+      expect(Object.values(runAttempts)).toHaveLength(3); // 3 runs
+      Object.values(runAttempts).forEach((attempts) => {
+        expect(attempts).toBe(3); // Each run attempted 3 times
+      });
+
+      // Verify execution order maintained retry attempts for each run
+      const run1Attempts = executedRuns.filter((r) => r.runId === "run1");
+      const run2Attempts = executedRuns.filter((r) => r.runId === "run2");
+      const run3Attempts = executedRuns.filter((r) => r.runId === "run3");
+
+      expect(run1Attempts).toHaveLength(3);
+      expect(run2Attempts).toHaveLength(3);
+      expect(run3Attempts).toHaveLength(3);
+
+      // Verify attempts are numbered correctly for each run
+      [run1Attempts, run2Attempts, run3Attempts].forEach((attempts) => {
+        expect(attempts.map((a) => a.attempt)).toEqual([1, 2, 3]);
+      });
+
+      // Verify no more retries occur
+      await setTimeout(1000);
+      expect(executedRuns).toHaveLength(9); // 3 runs * 3 attempts each
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("Should implement exponential backoff between retries", async ({ redisContainer }) => {
+    const executionTimes: number[] = [];
+    let startTime: number;
+
+    const minDelay = 100;
+    const factor = 2;
+
+    const queue = new ReleaseConcurrencyQueue<string>({
+      redis: {
+        keyPrefix: "release-queue:test:",
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+      },
+      executor: async (releaseQueue, runId) => {
+        const now = Date.now();
+        executionTimes.push(now);
+        console.log(`Execution at ${now - startTime}ms from start`);
+        throw new Error("Executor failed");
+      },
+      keys: {
+        fromDescriptor: (descriptor) => descriptor,
+        toDescriptor: (name) => name,
+      },
+      retry: {
+        maxRetries: 2,
+        backoff: {
+          minDelay,
+          maxDelay: 1000,
+          factor,
+        },
+      },
+      pollInterval: 50,
+    });
+
+    try {
+      startTime = Date.now();
+      await queue.attemptToRelease("test-queue", "run1", 1);
+
+      // Wait for all retries to complete
+      await setTimeout(1000);
+
+      // Should have 3 execution times (initial + 2 retries)
+      expect(executionTimes).toHaveLength(3);
+
+      const intervals = executionTimes.slice(1).map((time, i) => time - executionTimes[i]);
+      console.log("Intervals between retries:", intervals);
+
+      // First retry should be after ~200ms (minDelay + processing overhead)
+      const expectedFirstDelay = minDelay * 2; // Account for observed overhead
+      expect(intervals[0]).toBeGreaterThanOrEqual(expectedFirstDelay * 0.8);
+      expect(intervals[0]).toBeLessThanOrEqual(expectedFirstDelay * 1.5);
+
+      // Second retry should be after ~400ms (first delay * factor)
+      const expectedSecondDelay = expectedFirstDelay * factor;
+      expect(intervals[1]).toBeGreaterThanOrEqual(expectedSecondDelay * 0.8);
+      expect(intervals[1]).toBeLessThanOrEqual(expectedSecondDelay * 1.5);
+
+      // Log expected vs actual delays
+      console.log("Expected delays:", { first: expectedFirstDelay, second: expectedSecondDelay });
+    } finally {
+      await queue.quit();
+    }
+  });
+});

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

@@ -105,9 +105,7 @@ describe("RunEngine Waitpoints", () => {
         environmentId: authenticatedEnvironment.id,
         projectId: authenticatedEnvironment.project.id,
         organizationId: authenticatedEnvironment.organization.id,
-        releaseConcurrency: {
-          releaseQueue: true,
-        },
+        releaseConcurrency: true,
       });
       expect(result.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS");
       expect(result.runStatus).toBe("EXECUTING");

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

@@ -8,11 +8,11 @@ import { FairQueueSelectionStrategyOptions } from "../run-queue/fairQueueSelecti
 
 export type RunEngineOptions = {
   prisma: PrismaClient;
-  worker: WorkerConcurrencyOptions & {
+  worker: {
     redis: RedisOptions;
     pollIntervalMs?: number;
     immediatePollIntervalMs?: number;
-  };
+  } & WorkerConcurrencyOptions;
   machines: {
     defaultMachine: MachinePresetName;
     machines: Record<string, MachinePreset>;
@@ -35,6 +35,10 @@ export type RunEngineOptions = {
   heartbeatTimeoutsMs?: Partial<HeartbeatTimeouts>;
   queueRunsWaitingForWorkerBatchSize?: number;
   tracer: Tracer;
+  releaseConcurrency?: {
+    maxTokens?: number;
+    redis?: Partial<RedisOptions>;
+  };
 };
 
 export type HeartbeatTimeouts = {

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

@@ -589,11 +589,7 @@ export class RunQueue {
     );
   }
 
-  public async releaseConcurrency(
-    orgId: string,
-    messageId: string,
-    releaseForRun: boolean = false
-  ) {
+  public async releaseConcurrency(orgId: string, messageId: string) {
     return this.#trace(
       "releaseConcurrency",
       async (span) => {
@@ -617,7 +613,7 @@ export class RunQueue {
         return this.redis.releaseConcurrency(
           this.keys.messageKey(orgId, messageId),
           message.queue,
-          releaseForRun ? this.keys.currentConcurrencyKeyFromQueue(message.queue) : "",
+          this.keys.currentConcurrencyKeyFromQueue(message.queue),
           this.keys.envCurrentConcurrencyKeyFromQueue(message.queue),
           this.keys.projectCurrentConcurrencyKeyFromQueue(message.queue),
           this.keys.taskIdentifierCurrentConcurrencyKeyFromQueue(

internal-packages/run-engine/src/run-queue/keyProducer.ts

@@ -299,7 +299,35 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
       concurrencyKey: parts.at(9),
     };
   }
+  releaseConcurrencyKey(env: EnvDescriptor): string;
+  releaseConcurrencyKey(env: MinimalAuthenticatedEnvironment): string;
+  releaseConcurrencyKey(envOrDescriptor: EnvDescriptor | MinimalAuthenticatedEnvironment): string {
+    if ("id" in envOrDescriptor) {
+      return [
+        this.orgKeySection(envOrDescriptor.organization.id),
+        this.projKeySection(envOrDescriptor.project.id),
+        this.envKeySection(envOrDescriptor.id),
+        "release-concurrency",
+      ].join(":");
+    } else {
+      return [
+        this.orgKeySection(envOrDescriptor.orgId),
+        this.projKeySection(envOrDescriptor.projectId),
+        this.envKeySection(envOrDescriptor.envId),
+        "release-concurrency",
+      ].join(":");
+    }
+  }
 
+  releaseConcurrencyDescriptorFromQueue(queue: string): EnvDescriptor {
+    const parts = queue.split(":");
+
+    return {
+      orgId: parts[1],
+      projectId: parts[3],
+      envId: parts[5],
+    };
+  }
   private envKeySection(envId: string) {
     return `${constants.ENV_PART}:${envId}`;
   }

internal-packages/run-engine/src/run-queue/types.ts

@@ -96,6 +96,11 @@ export interface RunQueueKeyProducer {
   deadLetterQueueKey(env: MinimalAuthenticatedEnvironment): string;
   deadLetterQueueKey(env: EnvDescriptor): string;
   deadLetterQueueKeyFromQueue(queue: string): string;
+
+  releaseConcurrencyKey(env: MinimalAuthenticatedEnvironment): string;
+  releaseConcurrencyKey(env: EnvDescriptor): string;
+
+  releaseConcurrencyDescriptorFromQueue(queue: string): EnvDescriptor;
 }
 
 export type EnvQueues = {