WIP

Author: ericallam

apps/webapp/app/runEngine/concerns/queues.server.ts

@@ -27,6 +27,7 @@ export class DefaultQueueManager implements QueueManager {
   ): Promise<QueueProperties> {
     let queueName: string;
     let lockedQueueId: string | undefined;
+    let lockedQueueReleaseConcurrencyOnWaitpoint: boolean | undefined;
 
     // Determine queue name based on lockToVersion and provided options
     if (lockedBackgroundWorker) {
@@ -53,6 +54,9 @@ export class DefaultQueueManager implements QueueManager {
         // Use the validated queue name directly
         queueName = specifiedQueue.name;
         lockedQueueId = specifiedQueue.id;
+        lockedQueueReleaseConcurrencyOnWaitpoint =
+          typeof specifiedQueue.concurrencyLimit === "undefined" ||
+          specifiedQueue.releaseConcurrencyOnWaitpoint;
       } else {
         // No specific queue name provided, use the default queue for the task on the locked worker
         const lockedTask = await this.prisma.backgroundWorkerTask.findFirst({
@@ -91,6 +95,9 @@ export class DefaultQueueManager implements QueueManager {
         // Use the task's default queue name
         queueName = lockedTask.queue.name;
         lockedQueueId = lockedTask.queue.id;
+        lockedQueueReleaseConcurrencyOnWaitpoint =
+          typeof lockedTask.queue.concurrencyLimit === "undefined" ||
+          lockedTask.queue.releaseConcurrencyOnWaitpoint;
       }
     } else {
       // Task is not locked to a specific version, use regular logic
@@ -118,6 +125,7 @@ export class DefaultQueueManager implements QueueManager {
     return {
       queueName,
       lockedQueueId,
+      lockedQueueReleaseConcurrencyOnWaitpoint,
     };
   }
 

apps/webapp/app/runEngine/types.ts

@@ -52,6 +52,7 @@ export type QueueValidationResult =
 export type QueueProperties = {
   queueName: string;
   lockedQueueId?: string;
+  lockedQueueReleaseConcurrencyOnWaitpoint?: boolean;
 };
 
 export type LockedBackgroundWorker = Pick<

internal-packages/database/prisma/migrations/20250710163022_add_locked_queue_release_concurrency_to_task_run/migration.sql

@@ -0,0 +1,20 @@
+-- DropIndex
+DROP INDEX "SecretStore_key_idx";
+
+-- DropIndex
+DROP INDEX "TaskRun_runtimeEnvironmentId_createdAt_idx";
+
+-- DropIndex
+DROP INDEX "TaskRun_runtimeEnvironmentId_id_idx";
+
+-- AlterTable
+ALTER TABLE "TaskRun" ADD COLUMN     "lockedQueueReleaseConcurrencyOnWaitpoint" BOOLEAN;
+
+-- CreateIndex
+CREATE INDEX "SecretStore_key_idx" ON "SecretStore"("key" text_pattern_ops);
+
+-- CreateIndex
+CREATE INDEX "TaskRun_runtimeEnvironmentId_id_idx" ON "TaskRun"("runtimeEnvironmentId", "id" DESC);
+
+-- CreateIndex
+CREATE INDEX "TaskRun_runtimeEnvironmentId_createdAt_idx" ON "TaskRun"("runtimeEnvironmentId", "createdAt" DESC);

internal-packages/database/prisma/schema.prisma

@@ -555,9 +555,11 @@ model TaskRun {
   organizationId String?
 
   // The specific queue this run is in
-  queue         String
+  queue                                    String
   // The queueId is set when the run is locked to a specific queue
-  lockedQueueId String?
+  lockedQueueId                            String?
+  // Whether the locked queue specifies that concurrency should be released when a run is blocked by a waitpoint
+  lockedQueueReleaseConcurrencyOnWaitpoint Boolean?
 
   /// The main queue that this run is part of
   workerQueue String @default("main") @map("masterQueue")