feat(prefetch/faz3): per-source paralel rate limit (PL24/EMEX/pcat bağımsız)

Tek global 20/dk limiter tüm kaynakları paylaştırıyordu → yavaş pcat PL24/emex'i
açlığa düşürüyordu. Artık her kaynak Redis fixed-window ile BAĞIMSIZ kısılıyor
(SOURCE_RATE_MAX: pl24=20, emex=20, pcat=8; env PREFETCH_RATE_PL24/_EMEX/_PCAT).
Global WORKER_RATE_MAX güvenlik tavanı (20→60). RateLimitError yeni cause
"source-rate" → process() per-job defer eder (emex/pcat PL24'ü beklemez); cooldown
hâlâ tüm worker'ı duraklatır. PL24 default'u 20'de KALDI (ban riski; yüksek-rate
Faz 4 rotating proxy ile). +spec: under/over ceiling + unknown-source unlimited.

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

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

@@ -50,7 +50,7 @@ export function isLifecycleEmailEnabled(): boolean {
 export class RateLimitError extends Error {
   constructor(
     public readonly retryAfterMs: number,
-    public readonly cause: "cooldown" | "time-window",
+    public readonly cause: "cooldown" | "time-window" | "source-rate",
   ) {
     super(`Rate limited (${cause}) — retry after ${retryAfterMs}ms`);
     this.name = "RateLimitError";

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

@@ -29,6 +29,8 @@ function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
     exists: vi.fn(async (..._args: unknown[]) => false), // no cooldown / guard / complete marker
     get: vi.fn(async () => null), // no no-result residue
     set: vi.fn(async () => undefined),
+    incr: vi.fn(async (..._args: unknown[]) => 1), // per-source rate window counter
+    expire: vi.fn(async () => undefined),
     setNx: vi.fn(async () => true), // scan lock acquired by default
     getJson: vi.fn(async () => null), // Phase-2 cursor empty
     setJson: vi.fn(async () => undefined),
@@ -196,6 +198,30 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
     });
   });
 
+  describe("checkSourceRate — per-source rate limit", () => {
+    type CSR = { checkSourceRate: (s: string) => Promise<void> };
+
+    it("passes when under the source ceiling", async () => {
+      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
+      redis.incr.mockResolvedValueOnce(1);
+      await expect((service as never as CSR).checkSourceRate("pl24")).resolves.toBeUndefined();
+    });
+
+    it("throws a source-rate RateLimitError once over the ceiling", async () => {
+      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
+      redis.incr.mockResolvedValueOnce(21); // pl24 default ceiling is 20
+      await expect((service as never as CSR).checkSourceRate("pl24")).rejects.toMatchObject({
+        cause: "source-rate",
+      });
+    });
+
+    it("is unlimited (no counter) for a source without a configured ceiling", async () => {
+      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
+      await (service as never as CSR).checkSourceRate("unknown-source");
+      expect(redis.incr).not.toHaveBeenCalled();
+    });
+  });
+
   describe("queueCategoryJob — depth cap (no-op guard)", () => {
     // pl24 linkPath with no leaf marker (/bom/, /partinfo/ …) → treated as a
     // non-leaf folder that would normally queue a prefetch-children job.

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

@@ -72,8 +72,25 @@ const BACKFILL_SCAN_LOCK_TTL_S = 55 * 60;
  * cooldown still pauses the whole worker, so this only speeds up idle periods.
  */
 const WORKER_CONCURRENCY = Number(process.env.PREFETCH_CONCURRENCY) || 3;
-/** Global queue rate ceiling (jobs per minute). Was a hard 5/min. */
-const WORKER_RATE_MAX = Number(process.env.PREFETCH_RATE_MAX) || 20;
+/**
+ * Global queue rate ceiling (jobs/min) across ALL sources — now a SAFETY CAP,
+ * not the real throttle. Per-source limits (below) do the actual pacing so a
+ * slow/limited source can't starve the others. Defaults to the sum of the
+ * per-source ceilings + headroom. Was a hard 5/min, then a shared 20/min.
+ */
+const WORKER_RATE_MAX = Number(process.env.PREFETCH_RATE_MAX) || 60;
+/**
+ * Per-source rate ceilings (jobs/min). Each source is throttled INDEPENDENTLY
+ * (Redis fixed-window) so PL24, EMEX and parts-catalogs run concurrently at
+ * their own safe rates instead of fighting over one global budget. 0 = unlimited
+ * for that source. PL24 stays conservative (upstream ban risk); raise it only
+ * with rotating residential proxies. pcat is already paced by PCAT_PACE_MS.
+ */
+const SOURCE_RATE_MAX: Record<string, number> = {
+  pl24: Number(process.env.PREFETCH_RATE_PL24) || 20,
+  emex: Number(process.env.PREFETCH_RATE_EMEX) || 20,
+  "parts-catalogs": Number(process.env.PREFETCH_RATE_PCAT) || 8,
+};
 /**
  * Per-job pacing for parts-catalogs only (its browser/JWT capture is heavy).
  * Set 0 to disable. Other sources are paced by the limiter + cooldown alone.
@@ -136,7 +153,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     });
 
     this.logger.log(
-      `[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, ${WORKER_RATE_MAX} jobs/min, pcatPace=${PCAT_PACE_MS}ms)`,
+      `[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, global ${WORKER_RATE_MAX}/min, ` +
+        `per-source ${JSON.stringify(SOURCE_RATE_MAX)}, pcatPace=${PCAT_PACE_MS}ms)`,
     );
 
     // Hourly backfill scan — run IN-PROCESS, not as a BullMQ cron job. A cron
@@ -183,6 +201,16 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
   private async process(job: Job, token?: string): Promise<void> {
     try {
       const data = job.data as { source?: string };
+      // Per-source rate gate FIRST (before the pcat pace) so we don't burn the 15s
+      // sleep on a job we're about to defer. Scan jobs are exempt.
+      if (
+        data.source &&
+        (job.name === "prefetch-init" ||
+          job.name === "prefetch-children" ||
+          job.name === "prefetch-parts")
+      ) {
+        await this.checkSourceRate(data.source);
+      }
       if (data.source === "parts-catalogs" && PCAT_PACE_MS > 0) {
         await new Promise((r) => setTimeout(r, PCAT_PACE_MS));
       }
@@ -214,10 +242,11 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           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.
+        // time-window (off-hours, up to hours) and source-rate (this source hit
+        // its per-minute ceiling, ~sub-minute) both defer PER-JOB so other sources
+        // keep flowing — only cooldown pauses the whole worker.
         await job.moveToDelayed(Date.now() + delayMs, token);
-        this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (off-hours)`);
+        this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (${err.cause})`);
         throw new DelayedError();
       }
       throw err;
@@ -744,6 +773,25 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     return `prefetch:complete:${vehicleId}`;
   }
 
+  /**
+   * Per-source fixed-window (60s) rate limit. Increments the source's window
+   * counter; once it exceeds the source's ceiling, throws RateLimitError
+   * ("source-rate") so process() defers THIS job to the window's end while other
+   * sources keep flowing. 0 / unknown source ceiling = unlimited.
+   */
+  private async checkSourceRate(source: string): Promise<void> {
+    const max = SOURCE_RATE_MAX[source] ?? 0;
+    if (max <= 0) return;
+    const windowMs = 60_000;
+    const now = Date.now();
+    const key = `prefetch:rate:${source}:${Math.floor(now / windowMs)}`;
+    const n = await this.redis.incr(key);
+    if (n === 1) await this.redis.expire(key, 61);
+    if (n > max) {
+      throw new RateLimitError(windowMs - (now % windowMs) + 50, "source-rate");
+    }
+  }
+
   /** Record that a backfill attempt finished with the vehicle still at zero parts. */
   private async markNoResult(vehicleId: string): Promise<void> {
     const key = this.noResultKey(vehicleId);