Add configuration for the release concurrency queue

Author: ericallam

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

@@ -255,13 +255,16 @@ export class RunEngine {
         keyPrefix: `${options.queue.redis.keyPrefix}release-concurrency:`,
       },
       retry: {
-        maxRetries: 5, // TODO: Make this configurable
+        maxRetries: options.releaseConcurrency?.maxRetries ?? 5,
         backoff: {
-          minDelay: 1000, // TODO: Make this configurable
-          maxDelay: 10000, // TODO: Make this configurable
-          factor: 2, // TODO: Make this configurable
+          minDelay: options.releaseConcurrency?.backoff?.minDelay ?? 1000,
+          maxDelay: options.releaseConcurrency?.backoff?.maxDelay ?? 10000,
+          factor: options.releaseConcurrency?.backoff?.factor ?? 2,
         },
       },
+      consumersCount: options.releaseConcurrency?.consumersCount ?? 1,
+      pollInterval: options.releaseConcurrency?.pollInterval ?? 1000,
+      batchSize: options.releaseConcurrency?.batchSize ?? 10,
       executor: async (descriptor, runId) => {
         await this.#executeReleasedConcurrencyFromQueue(descriptor, runId);
       },

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

@@ -0,0 +1,10 @@
+import { containerTest } from "@internal/testcontainers";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+describe("RunEngine Releasing Concurrency", () => {
+  containerTest(
+    "blocking a run with a waitpoint with releasing concurrency",
+    async ({ prisma, redisOptions }) => {}
+  );
+});

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

@@ -38,6 +38,15 @@ export type RunEngineOptions = {
   releaseConcurrency?: {
     maxTokensRatio?: number;
     redis?: Partial<RedisOptions>;
+    maxRetries?: number;
+    consumersCount?: number;
+    pollInterval?: number;
+    batchSize?: number;
+    backoff?: {
+      minDelay?: number; // Defaults to 1000
+      maxDelay?: number; // Defaults to 60000
+      factor?: number; // Defaults to 2
+    };
   };
 };