internal-packages/run-engine/src/run-queue/index.ts (5)

358-366: Add explicit return type for clarity

Avoid implicit any/union drift across callers. Return type here is number.

-  public async getEnvConcurrencyBurstFactor(env: MinimalAuthenticatedEnvironment) {
+  public async getEnvConcurrencyBurstFactor(
+    env: MinimalAuthenticatedEnvironment
+  ): Promise<number> {

---

368-370: Type the SMEMBERS return

SMEMBERS returns string[]. Make it explicit to prevent accidental Set usage downstream.

-  public async getCurrentConcurrencyOfEnvironment(env: MinimalAuthenticatedEnvironment) {
+  public async getCurrentConcurrencyOfEnvironment(
+    env: MinimalAuthenticatedEnvironment
+  ): Promise<string[]> {

---

372-374: Type the SMEMBERS return

Same as above for queue scope.

-  public async getCurrentConcurrencyOfQueue(env: MinimalAuthenticatedEnvironment, queue: string) {
+  public async getCurrentConcurrencyOfQueue(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string
+  ): Promise<string[]> {

---

384-395: Minor: accept timestamp to avoid Date allocations in hot paths

Not urgent, but this helper may be used in loops. Accept a number to reduce allocations and string conversions.

-  public async lengthOfQueueAvailableMessages(
-    env: MinimalAuthenticatedEnvironment,
-    queue: string,
-    currentTime: Date = new Date(),
-    concurrencyKey?: string
-  ) {
-    return this.redis.zcount(
-      this.keys.queueKey(env, queue, concurrencyKey),
-      "-inf",
-      String(currentTime.getTime())
-    );
-  }
+  public async lengthOfQueueAvailableMessages(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string,
+    currentTimeMs: number = Date.now(),
+    concurrencyKey?: string
+  ): Promise<number> {
+    return this.redis.zcount(this.keys.queueKey(env, queue, concurrencyKey), "-inf", String(currentTimeMs));
+  }

---

453-459: Naming consistency: dequeued vs concurrency

Good addition. Consider documenting that “currentDequeued” is the user-facing metric whereas “currentConcurrency” is operational, to avoid confusion when both are exposed.

internal-packages/run-engine/src/engine/index.ts (4)

64-64: p-map default error behavior can abort whole repair

By default, p-map rejects on the first error (stopOnError: true). For admin repair flows, best-effort is preferable.

-    await pMap(
+    await pMap(
       completedRuns,
       async (run) => {
         await this.runQueue.acknowledgeMessage(run.orgId, run.id, {
           skipDequeueProcessing: true,
           removeFromWorkerQueue: false,
         });
       },
-      { concurrency: 5 }
+      { concurrency: 5, stopOnError: false }
     );

Repeat the same change in repairQueue.

---

1172-1209: Repair environment: swallow-and-log per-ack errors to avoid aborting all

Even with stopOnError: false, unexpected throw inside the mapper would still reject. Wrap the ack call to log and continue.

-    await pMap(
+    await pMap(
       completedRuns,
       async (run) => {
-        await this.runQueue.acknowledgeMessage(run.orgId, run.id, {
-          skipDequeueProcessing: true,
-          removeFromWorkerQueue: false,
-        });
+        try {
+          await this.runQueue.acknowledgeMessage(run.orgId, run.id, {
+            skipDequeueProcessing: true,
+            removeFromWorkerQueue: false, // consider making this configurable for deep cleanup
+          });
+        } catch (err) {
+          this.logger.error("repairEnvironment acknowledge failed", {
+            runId: run.id,
+            orgId: run.orgId,
+            error: err,
+          });
+        }
       },
-      { concurrency: 5 }
+      { concurrency: 5, stopOnError: false }
     );

---

1211-1251: Repair queue: same best-effort pattern and error handling

Mirror the defensive wrapping here too.

-    await pMap(
+    await pMap(
       completedRuns,
       async (run) => {
-        await this.runQueue.acknowledgeMessage(run.orgId, run.id, {
-          skipDequeueProcessing: true,
-          removeFromWorkerQueue: false,
-        });
+        try {
+          await this.runQueue.acknowledgeMessage(run.orgId, run.id, {
+            skipDequeueProcessing: true,
+            removeFromWorkerQueue: false,
+          });
+        } catch (err) {
+          this.logger.error("repairQueue acknowledge failed", {
+            queue,
+            runId: run.id,
+            orgId: run.orgId,
+            error: err,
+          });
+        }
       },
-      { concurrency: 5 }
+      { concurrency: 5, stopOnError: false }
     );

---

1646-1654: Outdated inline comment

The comment hardcodes “10 minutes” while completedAtOffsetMs is parameterized and is passed as 5000ms by the repair methods.

-          lte: new Date(Date.now() - completedAtOffsetMs), // This only finds runs that were completed more than 10 minutes ago
+          lte: new Date(Date.now() - completedAtOffsetMs), // Only runs completed earlier than completedAtOffsetMs ago

apps/webapp/app/routes/admin.api.v1.environments.$environmentId.engine.report.ts (4)

12-16: Validate and normalize query params (booleans/limits) to avoid negative skip/take

Prevent negative pagination and support common truthy values for verbose.

-const SearchParamsSchema = z.object({
-  verbose: z.string().default("0"),
-  page: z.coerce.number().optional(),
-  per_page: z.coerce.number().optional(),
-});
+const SearchParamsSchema = z.object({
+  verbose: z
+    .string()
+    .optional()
+    .transform((v) => v === "1" || v === "true"),
+  page: z.coerce.number().int().min(1).default(1),
+  per_page: z.coerce.number().int().min(1).max(500).default(50),
+});

---

66-68: Use sanitized defaults directly

Earlier schema change sets safe defaults and bounds.

-  const page = searchParams.page ?? 1;
-  const perPage = searchParams.per_page ?? 50;
+  const page = searchParams.page;
+  const perPage = searchParams.per_page;

---

88-92: Pass boolean verbose

Align with normalized schema.

-  const report = await engine.generateEnvironmentReport(
-    environment,
-    queues,
-    searchParams.verbose === "1"
-  );
+  const report = await engine.generateEnvironmentReport(environment, queues, !!searchParams.verbose);

---

57-61: determineEngineVersion might throw on missing worker version

Consider wrapping with try/catch and returning a 400 with the message to aid debugging, rather than a 500.

apps/webapp/app/routes/admin.api.v1.environments.$environmentId.engine.repair-queues.ts (5)

86-92: Make per-queue repairs resilient and abortable; capture errors and pass request.signal.

Today, a single rejection aborts the whole batch or bubbles a 500. Capture per-queue errors and tie work to the request’s abort signal.

Apply this diff:

-  const repairResults = await pMap(
-    queues,
-    async (queue) => {
-      return engine.repairQueue(environment, queue.name, parsedBody.dryRun);
-    },
-    { concurrency: 5 }
-  );
+  const repairResults = await pMap(
+    queues,
+    async (queue) => {
+      try {
+        const result = await engine.repairQueue(environment, queue.name, parsedBody.dryRun);
+        return { queue: queue.name, result };
+      } catch (err) {
+        return {
+          queue: queue.name,
+          error: err instanceof Error ? err.message : String(err),
+        };
+      }
+    },
+    { concurrency: 5, signal: request.signal }
+  );

---

84-85: Don’t fail the whole request if environment-level repair throws; return a structured error.

Keep the endpoint useful even if the environment repair step fails.

Apply this diff:

-  const repairEnvironmentResults = await engine.repairEnvironment(environment, parsedBody.dryRun);
+  let environmentResult: unknown;
+  try {
+    environmentResult = await engine.repairEnvironment(environment, parsedBody.dryRun);
+  } catch (err) {
+    environmentResult = { error: err instanceof Error ? err.message : String(err) };
+  }
@@
-  return json({ environment: repairEnvironmentResults, queues: repairResults });
+  return json({ environment: environmentResult, queues: repairResults });
 }

Also applies to: 94-95

---

46-51: Avoid fetching unused relations on environment (trim include).

organization and orgMember aren’t used. Reduce payload/CPU.

Apply this diff:

     include: {
-      organization: true,
       project: true,
-      orgMember: true,
     },

---

72-78: Only select queue fields you actually use.

You only need name for repairQueue; drop the rest to reduce transfer.

Apply this diff:

-    select: {
-      friendlyId: true,
-      name: true,
-      concurrencyLimit: true,
-      type: true,
-      paused: true,
-    },
+    select: { name: true },

---

66-82: Optional: surface unknown queue names to the caller.

If the request specifies queues that don’t exist, they’re silently ignored. Consider reporting a notFound list alongside queues so operators can catch typos.

Would you like me to add a small diff that computes requestedButMissing = parsedBody.queues.filter(q => !queues.some(x => x.name === q)) and returns it in the response?

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-08-29T10:06:49.293Z
Learning: Prefer Run Engine 2.0 via internal/run-engine; avoid extending legacy run engine code

Learnt from: matt-aitken
PR: triggerdotdev/trigger.dev#2264
File: apps/webapp/app/services/runsRepository.server.ts:172-174
Timestamp: 2025-07-12T18:06:04.133Z
Learning: In apps/webapp/app/services/runsRepository.server.ts, the in-memory status filtering after fetching runs from Prisma is intentionally used as a workaround for ClickHouse data delays. This approach is acceptable because the result set is limited to a maximum of 100 runs due to pagination, making the performance impact negligible.