Improved release concurrency accounting system + a sweeper to auto-refill tokens for snapshots that are no longer the latest snapshot on a run (e.g. the run has moved to a new snapshot state)

Author: ericallam

apps/webapp/app/env.server.ts

@@ -605,6 +605,11 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_RELEASE_CONCURRENCY_ENABLED: z.string().default("0"),
   RUN_ENGINE_RELEASE_CONCURRENCY_DISABLE_CONSUMERS: z.string().default("0"),
   RUN_ENGINE_RELEASE_CONCURRENCY_MAX_TOKENS_RATIO: z.coerce.number().default(1),
+  RUN_ENGINE_RELEASE_CONCURRENCY_RELEASINGS_MAX_AGE: z.coerce
+    .number()
+    .int()
+    .default(60_000 * 30),
+  RUN_ENGINE_RELEASE_CONCURRENCY_RELEASINGS_POLL_INTERVAL: z.coerce.number().int().default(60_000),
   RUN_ENGINE_RELEASE_CONCURRENCY_MAX_RETRIES: z.coerce.number().int().default(3),
   RUN_ENGINE_RELEASE_CONCURRENCY_CONSUMERS_COUNT: z.coerce.number().int().default(1),
   RUN_ENGINE_RELEASE_CONCURRENCY_POLL_INTERVAL: z.coerce.number().int().default(500),

apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts

@@ -155,6 +155,10 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
         label: "Release concurrency metadata",
         key: releaseConcurrencyMetadataKey,
       },
+      {
+        label: "Release concurrency releasings",
+        key: "engine:release-concurrency:releasings",
+      },
     ];
 
     return typedjson({

apps/webapp/app/v3/runEngine.server.ts

@@ -81,6 +81,8 @@ function createRunEngine() {
       disabled: env.RUN_ENGINE_RELEASE_CONCURRENCY_ENABLED === "0",
       disableConsumers: env.RUN_ENGINE_RELEASE_CONCURRENCY_DISABLE_CONSUMERS === "1",
       maxTokensRatio: env.RUN_ENGINE_RELEASE_CONCURRENCY_MAX_TOKENS_RATIO,
+      releasingsMaxAge: env.RUN_ENGINE_RELEASE_CONCURRENCY_RELEASINGS_MAX_AGE,
+      releasingsPollInterval: env.RUN_ENGINE_RELEASE_CONCURRENCY_RELEASINGS_POLL_INTERVAL,
       maxRetries: env.RUN_ENGINE_RELEASE_CONCURRENCY_MAX_RETRIES,
       consumersCount: env.RUN_ENGINE_RELEASE_CONCURRENCY_CONSUMERS_COUNT,
       pollInterval: env.RUN_ENGINE_RELEASE_CONCURRENCY_POLL_INTERVAL,

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

@@ -201,6 +201,9 @@ export class RunEngine {
 
     this.releaseConcurrencySystem = new ReleaseConcurrencySystem({
       resources,
+      maxTokensRatio: options.releaseConcurrency?.maxTokensRatio,
+      releasingsMaxAge: options.releaseConcurrency?.releasingsMaxAge,
+      releasingsPollInterval: options.releaseConcurrency?.releasingsPollInterval,
       queueOptions:
         typeof options.releaseConcurrency?.disabled === "boolean" &&
         options.releaseConcurrency.disabled
@@ -223,33 +226,6 @@ export class RunEngine {
               consumersCount: options.releaseConcurrency?.consumersCount ?? 1,
               pollInterval: options.releaseConcurrency?.pollInterval ?? 1000,
               batchSize: options.releaseConcurrency?.batchSize ?? 10,
-              executor: async (descriptor, snapshotId) => {
-                return await this.releaseConcurrencySystem.executeReleaseConcurrencyForSnapshot(
-                  snapshotId
-                );
-              },
-              maxTokens: async (descriptor) => {
-                const environment = await this.prisma.runtimeEnvironment.findFirstOrThrow({
-                  where: { id: descriptor.envId },
-                  select: {
-                    maximumConcurrencyLimit: true,
-                  },
-                });
-
-                return (
-                  environment.maximumConcurrencyLimit *
-                  (options.releaseConcurrency?.maxTokensRatio ?? 1.0)
-                );
-              },
-              keys: {
-                fromDescriptor: (descriptor) =>
-                  `org:${descriptor.orgId}:proj:${descriptor.projectId}:env:${descriptor.envId}`,
-                toDescriptor: (name) => ({
-                  orgId: name.split(":")[1],
-                  projectId: name.split(":")[3],
-                  envId: name.split(":")[5],
-                }),
-              },
               tracer: this.tracer,
             },
     });