fix(api): pause whole worker on cooldown, only per-job defer for off-hours

Per-job moveToDelayed for cooldown livelocked the worker: the activity key is
refreshed by every user request, so 60s later the deferred job comes back, key
is still set, defers again. Last 3h on prod logged ~1800 deferrals against 12
real inits and one completion every ~6 min.

Tag RateLimitError with a `cause`. checkCooldown throws "cooldown"; the worker
now calls `this.worker.rateLimit(delayMs)` and throws Worker.RateLimitError() —
the whole queue waits once instead of cycling every job. checkTimeWindow throws
"time-window"; that branch keeps the existing job.moveToDelayed (per-job) so
EMEX (no scrape window) keeps flowing while PL24/pcat jobs sleep till 09:00.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/api/src/jobs/prefetch-utils.ts

@@ -19,12 +19,23 @@ export function isCatalogBackfillEnabled(): boolean {
 }
 
 /**
- * Custom error that tells BullMQ to retry after a delay.
- * The worker catches this and re-queues the job with the specified delay.
+ * Signals the worker that this job is rate-limited and should be deferred.
+ *
+ * `cause` distinguishes the deferral strategy:
+ *  - "cooldown" — active user on the source. The cooldown key is refreshed by
+ *    every user request, so per-job 60s defer livelocks: the job comes back,
+ *    cooldown is still set, defers again. Handled by pausing the WHOLE worker
+ *    for retryAfterMs (BullMQ `worker.rateLimit()`), so the queue waits once
+ *    instead of cycling.
+ *  - "time-window" — off-hours; only PL24/parts-catalogs are affected. Use
+ *    per-job moveToDelayed so EMEX jobs (no window) keep flowing.
  */
 export class RateLimitError extends Error {
-  constructor(public readonly retryAfterMs: number) {
-    super(`Rate limited — retry after ${retryAfterMs}ms`);
+  constructor(
+    public readonly retryAfterMs: number,
+    public readonly cause: "cooldown" | "time-window",
+  ) {
+    super(`Rate limited (${cause}) — retry after ${retryAfterMs}ms`);
     this.name = "RateLimitError";
   }
 }
@@ -38,7 +49,7 @@ export async function checkCooldown(redis: RedisService, source: string): Promis
   const exists = await redis.exists(key);
   if (exists) {
     const retryMs = source === "parts-catalogs" ? 120_000 : 60_000;
-    throw new RateLimitError(retryMs);
+    throw new RateLimitError(retryMs, "cooldown");
   }
 }
 
@@ -70,7 +81,7 @@ export function isWithinTimeWindow(source: string): boolean {
  */
 export function checkTimeWindow(source: string): void {
   if (!isWithinTimeWindow(source)) {
-    throw new RateLimitError(msUntilNext9AM());
+    throw new RateLimitError(msUntilNext9AM(), "time-window");
   }
 }
 

apps/api/src/jobs/prefetch-worker.service.ts

@@ -103,13 +103,23 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       }
       this.logger.warn(`[prefetch] Unknown job name: ${job.name}`);
     } catch (err) {
-      // Rate-limited (active user cooldown or off-hours window): defer the job to
-      // its requested time WITHOUT consuming a retry attempt. moveToDelayed +
-      // DelayedError is BullMQ's contract for "not done, not failed — try later".
       if (err instanceof RateLimitError) {
         const delayMs = Math.max(1000, err.retryAfterMs);
+        if (err.cause === "cooldown") {
+          // Pause the WHOLE worker for the cooldown window: the per-source
+          // activity key is refreshed by every user request, so per-job defer
+          // livelocks (job comes back, key is still set, defer again, …). One
+          // global wait is correct: when it ends and cooldown is still active,
+          // we'll pause again — at most ~1 cycle per retryAfterMs instead of
+          // cycling every queued job through.
+          await this.worker?.rateLimit(delayMs);
+          this.logger.debug(`[prefetch] Worker paused ${delayMs}ms (cooldown ${job.data.source})`);
+          throw Worker.RateLimitError();
+        }
+        // Time-window deferral can be hours (msUntilNext9AM). Per-job defer so
+        // EMEX (no window) keeps flowing while PL24/pcat jobs sleep till 09:00.
         await job.moveToDelayed(Date.now() + delayMs, token);
-        this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (rate-limited)`);
+        this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (off-hours)`);
         throw new DelayedError();
       }
       throw err;