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

112-115: Cooloff state maps: pruning strategy and lifecycle

• Clearing both maps every 10 minutes resets active cooloffs and counters, which can re-expose saturated queues sooner than intended. Prefer pruning only expired cooloffs while keeping counters (or age-based pruning) to preserve behavior and still bound growth.
• The interval is never cleared. If multiple MarQS instances ever get created (tests, hot reloads), this can leak timers.

Prune precisely-expired entries and keep counts:

- this.clearCooloffPeriodInterval = setInterval(() => {
-   this.queueDequeueCooloffCounts.clear();
-   this.queueDequeueCooloffPeriod.clear();
- }, 60_000 * 10); // 10 minutes
+ this.clearCooloffPeriodInterval = setInterval(() => {
+   const now = Date.now();
+   for (const [queue, until] of this.queueDequeueCooloffPeriod) {
+     if (until <= now) this.queueDequeueCooloffPeriod.delete(queue);
+   }
+   // Optional: cap counts map size
+   if (this.queueDequeueCooloffCounts.size > 50_000) {
+     this.queueDequeueCooloffCounts.clear();
+   }
+ }, 600_000); // 10 minutes

And add a small disposer to avoid timer leaks:

public dispose() {
  clearInterval(this.clearCooloffPeriodInterval);
}

(You can call dispose() from your app shutdown hook if available.)

Also applies to: 124-129

---

844-857: Use .has() to check map membership and avoid truthiness pitfalls

The current truthy check would skip a 0 timestamp (unlikely here), and it reads clearer to check existence then compare.

Apply:

- const cooloffPeriod = this.queueDequeueCooloffPeriod.get(messageQueue);
- // If the queue is in a cooloff period, skip attempting to dequeue from it
- if (cooloffPeriod) {
-   // If the cooloff period is still active, skip attempting to dequeue from it
-   if (cooloffPeriod > Date.now()) {
+ const cooloffUntil = this.queueDequeueCooloffPeriod.get(messageQueue);
+ if (cooloffUntil !== undefined) {
+   if (cooloffUntil > Date.now()) {
      coolOffPeriodCount++;
      continue;
-   } else {
-     // If the cooloff period is over, delete the cooloff period and attempt to dequeue from the queue
-     this.queueDequeueCooloffPeriod.delete(messageQueue);
-   }
- }
+   }
+   // Cooloff period over; remove and attempt dequeue
+   this.queueDequeueCooloffPeriod.delete(messageQueue);
+ }

---

967-968: Expose threshold/period in telemetry to aid tuning

You already expose cooloff_period_count. Add the active threshold and period to spans so we can correlate tuning with effect.

Example:

 span.setAttribute("attempted_envs", attemptedEnvs);
 span.setAttribute("message_count", messageCount);
 span.setAttribute("cooloff_period_count", coolOffPeriodCount);
+span.setAttribute("cooloff_threshold", Math.max(10, this.options.sharedWorkerQueueCooloffCountThreshold ?? 10));
+span.setAttribute("cooloff_period_ms", this.options.sharedWorkerQueueCooloffPeriodMs ?? 10_000);

---

844-910: (Optional) Avoid cooling off queues that are about to become available

Empty dequeues can be due to:

• hard concurrency caps (intended to cool off), or
• scheduled messages just a few hundred ms in the future.

Cooling off the latter can delay near-ready work by the full cooloff period. If practical, consider checking the oldest score (e.g., zrange queueKey WITHSCORES 0 0) and only cool off if (oldestScore - now) exceeds a small grace window (e.g., > 2s). This is an extra Redis call, so apply behind a feature flag or only after N consecutive empties.

Happy to sketch a variant that consults oldestScore with a bounded cost and caches it for the scan pass.

---

835-910: Cluster-wide cooldown semantics (optional)

The cooldown state is process-local. If multiple webapp instances run the consumer, each will independently probe the same saturated queues. If this becomes an issue at scale, move the cooldown to Redis (e.g., SET queue:cooloff: with PX and NX) so all consumers respect it.

I can provide a minimal Redis-based design (SETNX + TTL) that keeps the same behavior with negligible overhead.

---

81-83: Naming nit: “cooldown” vs “cooloff”

We use “cooldown” elsewhere in infra contexts; consider standardizing. Variable naming is currently mixed (coolOffPeriodCount vs queueDequeueCooloffPeriod). Purely cosmetic.

Also applies to: 844-910

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 (3)

46-46: Good switch to timers/promises setInterval

Importing setInterval from node:timers/promises enables an abortable, non-overlapping loop. Nice.

---

751-757: Non-overlapping consumer loop is safer

for-await over setIntervalAsync with an AbortSignal prevents overlapping ticks. This should reduce contention and surprise concurrency.

---

915-916: Resetting the counter on success is correct

Clearing the per-queue counter on successful dequeue prevents inadvertent cooloffs after activity resumes.