apps/webapp/app/env.server.ts (1)

416-423: Optional: consider BoolEnv for the two boolean-like flags for consistency.

Most toggles here are "0"/"1" strings, but a subset uses BoolEnv. If you want stricter typing and easier consumption, you could switch MARQS_SHARED_WORKER_QUEUE_EAGER_DEQUEUE_ENABLED and MARQS_WORKER_ENABLED to BoolEnv and adjust call sites. If maintaining the existing "0"/"1" convention is intentional, ignore this.

apps/webapp/app/v3/marqs/index.server.ts (4)

47-50: Use named zod import for consistency with the codebase.

Other files in this repo/webapp import as import { z } from "zod". Align this to reduce friction.

-import z from "zod";
+import { z } from "zod";

---

136-138: Optional: expose a stop/close hook.

You start the worker when enabled; consider exposing a stop() that calls this.worker.stop() to support graceful shutdowns in tests or process tear-down.

---

933-990: Consider draining more per job to reduce scheduling overhead.

Current job enqueues itself again if it hit maxCount, with a fixed 500ms delay. For hot queues this increases Redis ops and latency. Optionally: loop until fewer than maxCount dequeued or a small time budget is exceeded.

Sketch:

-      // If we dequeued the max count, we need to enqueue another job to dequeue the next batch
-      if (dequeuedMessages.length === maxCount) {
+      // If we dequeued the max count, we need to enqueue another job to dequeue the next batch
+      if (dequeuedMessages.length === maxCount) {
         await this.worker.enqueueOnce({
           id: queueKey,
           job: "processQueueForWorkerQueue",
           payload: {
             queueKey,
             parentQueueKey,
           },
-          availableAt: new Date(Date.now() + 500), // 500ms from now
+          availableAt: new Date(Date.now() + EAGER_DEQUEUE_DEBOUNCE_MS),
         });
       }

---

117-135: Optional: add lightweight instrumentation around enqueueOnce failures.

If Redis is transiently unavailable, these eager-dequeue nudges will be dropped silently. Consider wrapping in a small try/catch with a warn-level log and maybe a counter metric.

Also applies to: 328-341, 933-990, 1022-1034, 2617-2629

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-07-18T17:49:47.180Z
Learning: Applies to apps/webapp/**/*.{ts,tsx} : In the webapp, all environment variables must be accessed through the `env` export of `env.server.ts`, instead of directly accessing `process.env`.

apps/webapp/app/v3/marqs/index.server.ts (4)

80-88: Worker options shape looks good.

The added eagerDequeuingEnabled and workerOptions provide a clean feature toggle and tuning surface. No issues.

---

90-98: Catalog entry is well-typed and scoped.

Defining processQueueForWorkerQueue in a local catalog with a 30s visibility timeout makes sense for this workload.

---

117-135: No remaining direct process.env.MARQS_ references found

A full Ripgrep scan across all .ts and .tsx files did not surface any direct process.env.MARQS_… usages, confirming that all environment‐flag access has been properly routed through the env module.

---

1022-1034: Gate ack-triggered eager enqueue on worker enabled & extract debounce constant

We should avoid scheduling jobs when the worker is disabled, just as we do on the enqueue path, and replace the magic “500” ms debounce with a named constant.

• Define a new constant in apps/webapp/app/v3/marqs/constants.server.ts

 export const MARQS_SCHEDULED_REQUEUE_AVAILABLE_AT_THRESHOLD_IN_MS = /*...*/;
+export const MARQS_EAGER_DEQUEUE_DEBOUNCE_IN_MS = 500;

• Update the ack‐triggered eager‐enqueue at lines 1024–1034 of index.server.ts

-        if (this.options.eagerDequeuingEnabled && message.parentQueue === sharedQueueKey) {
+        if (
+          this.options.eagerDequeuingEnabled &&
+          this.options.workerOptions?.enabled &&
+          message.parentQueue === sharedQueueKey
+        ) {
           await this.worker.enqueueOnce({
             id: message.queue,
             job: "processQueueForWorkerQueue",
             payload: {
               queueKey: message.queue,
               parentQueueKey: message.parentQueue,
             },
-            availableAt: new Date(Date.now() + 500), // 500ms from now
+            availableAt: new Date(Date.now() + MARQS_EAGER_DEQUEUE_DEBOUNCE_IN_MS),
           });
         }

• Apply the same guard and constant at the reservation path (around line 328):

-        if (env.type !== "DEVELOPMENT" && this.options.eagerDequeuingEnabled) {
+        if (
+          env.type !== "DEVELOPMENT" &&
+          this.options.eagerDequeuingEnabled &&
+          this.options.workerOptions?.enabled
+        ) {
           // …
           availableAt: new Date(Date.now() + MARQS_EAGER_DEQUEUE_DEBOUNCE_IN_MS),
         }

apps/webapp/app/env.server.ts (1)

416-423: Split “workers” from “limit” and align env var naming with existing conventions.

You’ve added a distinct workers count (good), but it’s named MARQS_WORKER_COUNT, which diverges from the pattern used elsewhere (e.g., COMMON_WORKER_CONCURRENCY_WORKERS, ADMIN_WORKER_CONCURRENCY_WORKERS, etc.). Introduce MARQS_WORKER_CONCURRENCY_WORKERS (with a default that falls back to MARQS_WORKER_COUNT for backward compatibility) and wire the consumer to use it. This prevents accidental coupling between “workers” and “limit” and keeps naming consistent across the codebase.

Apply in this file:

   MARQS_SHARED_WORKER_QUEUE_EAGER_DEQUEUE_ENABLED: z.string().default("0"),
   MARQS_WORKER_ENABLED: z.string().default("0"),
-  MARQS_WORKER_COUNT: z.coerce.number().int().default(2),
+  MARQS_WORKER_COUNT: z.coerce.number().int().default(2), // legacy alias
+  // Preferred name to match existing convention (e.g., *_WORKER_CONCURRENCY_WORKERS)
+  MARQS_WORKER_CONCURRENCY_WORKERS: z.coerce
+    .number()
+    .int()
+    .default(process.env.MARQS_WORKER_COUNT ? parseInt(process.env.MARQS_WORKER_COUNT) : 2),
   MARQS_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
   MARQS_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(5),
   MARQS_WORKER_POLL_INTERVAL_MS: z.coerce.number().int().default(100),
   MARQS_WORKER_IMMEDIATE_POLL_INTERVAL_MS: z.coerce.number().int().default(100),
   MARQS_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),

Follow-up: update the consumer in apps/webapp/app/v3/marqs/index.server.ts to read env.MARQS_WORKER_CONCURRENCY_WORKERS first, with a fallback to env.MARQS_WORKER_COUNT.

Run this to find any other MARQS worker vars that should follow the convention:

#!/bin/bash
# Inspect MARQS worker env naming consistency
rg -n --color=never -C2 -e 'MARQS_.*WORKER.*(COUNT|CONCURRENCY_WORKERS|CONCURRENCY_LIMIT|TASKS_PER_WORKER)'

apps/webapp/app/v3/marqs/index.server.ts (2)

2616-2636: Normalize worker keyPrefix and prefer the new workers var with legacy fallback.

• Worker keys share KEY_PREFIX with core MarQS keys; add “:worker” to isolate the worker’s namespace and avoid potential collisions.
• Read MARQS_WORKER_CONCURRENCY_WORKERS first (consistent naming), falling back to MARQS_WORKER_COUNT for backward compatibility.

     workerOptions: {
       enabled: env.MARQS_WORKER_ENABLED === "1",
       pollIntervalMs: env.MARQS_WORKER_POLL_INTERVAL_MS,
       immediatePollIntervalMs: env.MARQS_WORKER_IMMEDIATE_POLL_INTERVAL_MS,
       shutdownTimeoutMs: env.MARQS_WORKER_SHUTDOWN_TIMEOUT_MS,
       concurrency: {
-        workers: env.MARQS_WORKER_COUNT,
+        workers: env.MARQS_WORKER_CONCURRENCY_WORKERS ?? env.MARQS_WORKER_COUNT,
         tasksPerWorker: env.MARQS_WORKER_CONCURRENCY_TASKS_PER_WORKER,
         limit: env.MARQS_WORKER_CONCURRENCY_LIMIT,
       },
       redisOptions: {
-        keyPrefix: KEY_PREFIX,
+        // ensure worker keys are isolated and avoid "marqs::worker"
+        keyPrefix: `${KEY_PREFIX.replace(/:$/, "")}:worker`,
         port: env.REDIS_PORT ?? undefined,
         host: env.REDIS_HOST ?? undefined,
         username: env.REDIS_USERNAME ?? undefined,
         password: env.REDIS_PASSWORD ?? undefined,
         enableAutoPipelining: true,
         ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
       },
     },

Quick checks:

#!/bin/bash
# 1) Confirm we don't already use a dedicated marqs worker prefix
rg -n -C2 -F 'marqs::worker' || true
rg -n -C2 -F ':worker' apps/webapp/app/v3/marqs/index.server.ts

# 2) Verify consumers of workers vs limit
rg -n -C2 'MARQS_WORKER_(COUNT|CONCURRENCY_WORKERS|CONCURRENCY_LIMIT|TASKS_PER_WORKER)' apps/webapp/app

# 3) Ensure all 500ms occurrences are replaced
rg -n -C2 'Date\\.now\\(\\) \\+ 500' apps/webapp/app/v3/marqs/index.server.ts

---

1021-1034: Avoid enqueueing worker jobs when the worker is disabled; reuse debounce constant.

Prevents job backlog if the worker is off, and removes another magic 500.

-        if (this.options.eagerDequeuingEnabled && message.parentQueue === sharedQueueKey) {
+        if (
+          this.options.eagerDequeuingEnabled &&
+          this.options.workerOptions?.enabled &&
+          message.parentQueue === sharedQueueKey
+        ) {
           await this.worker.enqueueOnce({
             id: message.queue,
             job: "processQueueForWorkerQueue",
             payload: {
               queueKey: message.queue,
               parentQueueKey: message.parentQueue,
             },
-            availableAt: new Date(Date.now() + 500), // 500ms from now
+            availableAt: new Date(Date.now() + EAGER_DEQUEUE_DEBOUNCE_MS),
           });
         }

apps/webapp/app/v3/marqs/index.server.ts (1)

932-988: Use the same debounce constant when chaining batches.

Keep the re-enqueue delay consistent with the enqueue path.

       // If we dequeued the max count, we need to enqueue another job to dequeue the next batch
       if (dequeuedMessages.length === maxCount) {
         await this.worker.enqueueOnce({
           id: queueKey,
           job: "processQueueForWorkerQueue",
           payload: {
             queueKey,
             parentQueueKey,
           },
-          availableAt: new Date(Date.now() + 500), // 500ms from now
+          availableAt: new Date(Date.now() + EAGER_DEQUEUE_DEBOUNCE_MS),
         });
       }

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

apps/webapp/app/v3/marqs/index.server.ts (1)

119-137: Worker boot lifecycle looks good.

Worker is constructed with catalog and only started when enabled; jobs route to a private method with schema-validated payload. No issues spotted here.