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

177-177: Consider documenting the rationale for the 1-second timeout.

The reduction from 10 seconds to 1 second might affect trace collection completeness. While this change likely improves responsiveness, consider adding a comment explaining why 1 second is now sufficient.

-      traceTimeoutSeconds: options.traceTimeoutSeconds ?? 1,
+      // Reduced from 10s to 1s to improve trace responsiveness with the new dequeue strategy
+      traceTimeoutSeconds: options.traceTimeoutSeconds ?? 1,

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

413-414: Document the performance implications of these settings.

The new environment variables control critical performance characteristics of the dequeue system. Consider adding comments to explain:

1. The trade-off between interval frequency and CPU usage
2. The impact of max message count on throughput vs latency

+  // Interval between worker queue consumer iterations (ms). Lower values increase throughput but use more CPU.
   MARQS_SHARED_WORKER_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(250),
+  // Maximum messages to dequeue per iteration. Higher values improve throughput but may increase latency.
   MARQS_SHARED_WORKER_QUEUE_MAX_MESSAGE_COUNT: z.coerce.number().int().default(10),

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

676-734: Consider adding metrics for monitoring consumer health.

The background consumer implementation is solid, but consider adding:

1. Metrics for successful vs failed iterations
2. Alerts if the consumer stops unexpectedly
3. Configurable backoff on repeated failures

 async #startSharedWorkerQueueConsumer(consumerId: string, abortController: AbortController) {
   let lastProcessedAt = Date.now();
   let processedCount = 0;
+  let consecutiveErrors = 0;
+  const MAX_CONSECUTIVE_ERRORS = 10;

   try {
     for await (const _ of setInterval(
       this.options.sharedWorkerQueueConsumerIntervalMs ?? 500,
       null,
       {
         signal: abortController.signal,
       }
     )) {
       // ... existing code ...

       const [error, results] = await tryCatch(this.#processSharedWorkerQueue(consumerId));

       if (error) {
+        consecutiveErrors++;
         logger.error(`Failed to process shared worker queue`, {
           error,
           service: this.name,
           consumerId,
+          consecutiveErrors,
         });

+        if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) {
+          logger.error(`Too many consecutive errors, stopping consumer`, {
+            consumerId,
+            consecutiveErrors,
+          });
+          throw new Error(`Consumer stopped after ${consecutiveErrors} consecutive errors`);
+        }

         continue;
       }

+      consecutiveErrors = 0; // Reset on success
       // ... rest of the code ...
     }
   } catch (error) {
     // ... existing error handling ...
   }
 }

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/types.ts (1)

39-39: LGTM! Interface extension is well-designed.

The addition of sharedWorkerQueueKey(): string follows the established pattern and integrates cleanly with the existing key producer interface.

apps/webapp/app/v3/marqs/marqsKeyProducer.ts (2)

5-5: LGTM! Constant follows existing naming conventions.

The SHARED_WORKER_QUEUE constant is appropriately placed and follows the established naming pattern.

---

112-114: LGTM! Clean implementation of the interface requirement.

The method correctly returns the constant and maintains consistency with other similar methods in the class.

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

169-169: LGTM! Proper cleanup function storage.

The addition of _stopWorkerQueueConsumer follows the established pattern for managing lifecycle functions.

---

260-262: LGTM! Proper initialization and logging.

The consumer startup logic correctly stores the stop function and provides helpful console output.

---

440-440: API rename is consistent with the new architecture.

The method name change from dequeueMessageInSharedQueue to dequeueMessageFromSharedWorkerQueue better reflects the new dequeue strategy.

---

237-239: No additional cleanup needed for startup errors

Our investigation shows that marqs.startSharedWorkerQueueConsumer always returns a valid “stop” callback synchronously and internally catches any asynchronous startup errors:

• In index.server.ts, startSharedWorkerQueueConsumer constructs an AbortController, kicks off the async loop (catching and logging any non-AbortError exceptions), and then immediately returns the cleanup function () => abortController.abort()—so even if the consumer loop fails, the abort callback remains available.
• In SharedQueueConsumer#enable (sharedQueueConsumer.server.ts), _stopWorkerQueueConsumer is set to that callback before any work begins, and stop() always invokes it before resetting _enabled = false, ensuring proper teardown.

Since there are no synchronous failure paths that bypass assignment of _stopWorkerQueueConsumer, and all async errors are logged and don’t leave the consumer running, no extra error-handling or state cleanup is required.

Likely an incorrect or invalid review comment.

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

45-46: LGTM! Proper imports for the new async functionality.

The addition of setInterval from node:timers/promises and tryCatch utilities supports the new background consumer implementation.

---

75-76: LGTM! Options properly integrated into MarQS configuration.

The new options for controlling the shared worker queue consumer are well-placed and follow the existing naming conventions.

---

495-499: Consider handling edge cases in batch dequeue.

When changing from single to batch dequeue, ensure that:

1. Empty arrays are handled correctly
2. The transition maintains backward compatibility

---

566-658: Well-structured implementation of the new dequeue strategy.

The new dequeueMessageFromSharedWorkerQueue method:

• Properly traces operations with OpenTelemetry
• Correctly handles the two-step process (pop from worker queue, then read message)
• Maintains compatibility with the subscriber pattern
• Includes proper error handling

---

660-674: Robust consumer lifecycle management.

The startSharedWorkerQueueConsumer method properly:

• Uses AbortController for clean shutdown
• Handles errors without crashing
• Returns a cleanup function

---

739-871: Excellent implementation of the batch processing logic.

The #processSharedWorkerQueue method:

• Efficiently batches messages from multiple environments
• Properly traces all operations
• Handles errors gracefully per queue
• Uses RPUSH for O(1) insertion into the worker queue

---

1478-1550: Clean adaptation to support batch dequeuing.

The changes to #callDequeueMessages:

• Properly handle the maxCount parameter
• Correctly parse the flattened array response from Redis
• Maintain backward compatibility with single message dequeue (maxCount=1)

---

1980-2062: Well-designed Lua script for batch dequeue operations.

The dequeueMessages Lua script efficiently:

• Calculates available capacity across both queue and environment limits
• Atomically dequeues up to the calculated limit
• Maintains all concurrency tracking correctly
• Returns results in a flat array format for efficiency

---

2064-2082: Simple and efficient worker queue pop implementation.

The popMessageFromWorkerQueue Lua script correctly implements LPOP with queue length tracking.

---

2327-2347: TypeScript declarations properly updated.

The ioredis module augmentation correctly reflects the new method signatures and return types.

---

2483-2484: Configuration properly wired from environment variables.

The new options are correctly passed from environment variables to the MarQS configuration.