v3: Upgrades to MarQS (#989)

* Upgrades to MarQS

- It’s “queue choosing” algorithm is now MUCH better (pretty much didn’t work before) and should be performant even when we have a bunch of prod queues
- There are now concurrency limits at the environment and organization level, as well as the task/queue level. So if any of them are at capacity the message won’t be dequeued. This means we can have an org wide concurrency limit, as well as different limits for dev/prod/staging
- I’ve added an admin API that can be used to update the org/env concurrency limits

* More MarQS upgrades

- Extract the queue priortity choosing strategy into an interface
- Implement a much better weighted average strategy
- “Slide the window” of queue candidates if the parent queue sends all at-capacity queues (so we won’t get stuck attempting to choose the same 12 full queues)
- Added some unit tests for the priority stuff
- Added some ideas for expanding the priority choosing strategy with more dynamic features

* Fixes an issue with the shared queue consumer relying on the queue name to get the env id

Author: ericallam

apps/webapp/app/env.server.ts

@@ -63,6 +63,8 @@ const EnvironmentSchema = z.object({
   REDIS_PASSWORD: z.string().optional(),
   REDIS_TLS_DISABLED: z.string().optional(),
 
+  DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(5),
+  DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),
 

apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts

@@ -0,0 +1,66 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+const ParamsSchema = z.object({
+  environmentId: z.string(),
+});
+
+const RequestBodySchema = z.object({
+  envMaximumConcurrencyLimit: z.number(),
+  orgMaximumConcurrencyLimit: z.number(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  const parsedParams = ParamsSchema.parse(params);
+
+  const rawBody = await request.json();
+  const body = RequestBodySchema.parse(rawBody);
+
+  const environment = await prisma.runtimeEnvironment.update({
+    where: {
+      id: parsedParams.environmentId,
+    },
+    data: {
+      maximumConcurrencyLimit: body.envMaximumConcurrencyLimit,
+      organization: {
+        update: {
+          data: {
+            maximumConcurrencyLimit: body.orgMaximumConcurrencyLimit,
+          },
+        },
+      },
+    },
+    include: {
+      organization: true,
+      project: true,
+    },
+  });
+
+  await marqs?.updateEnvConcurrencyLimits(environment);
+
+  return json({ success: true });
+}

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

@@ -13,7 +13,7 @@ import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { CancelAttemptService } from "../services/cancelAttempt.server";
 import { CompleteAttemptService } from "../services/completeAttempt.server";
 import { attributesFromAuthenticatedEnv } from "../tracer.server";