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

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.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?.updateGlobalConcurrencyLimits(environment);
+
+  return json({ success: true });
+}