fix(prefetch): Phase-2 kilidini aç (pressure vs total gate) + tamamlanma muhasebesi

Sorun (2026-08-01): wait=0/active=0 iken scan backlog=11495 hesaplıyordu — hepsi
günlük-bütçe ile ertesi gün 00:00'a park edilmiş pcat job'ı. backlog>maxBacklog
olduğu için Phase-2 günün çoğunda askıdaydı; 2143 kısmi araç ilerlemiyordu.

Adversarial analiz kritik uyarı verdi: delayed'ı gate'ten çıkarmak TEK BAŞINA
470k runaway'i tekrarlar (üretim ~288k job/gün vs bütçe 65k/gün; bütçe-defer'leri
skipAttempt ile attempts tüketmediğinden asla düşmez). Bu yüzden üretim kaynağında
kesiliyor + ikinci tavan ekleniyor:

- Gate ikiye ayrıldı: pressure (waiting+active+10dk içinde vadesi gelen delayed,
  HER İKİ kuyruk) vs maxBacklog; total (tüm bekleyen) vs HARD_MAX_TOTAL_JOBS=50k.
  zcount BullMQ delayed ZSET score'u (dueMs*4096+seq) ile; hata halinde fail-CLOSED.
- ÜRETİMİ KES: scan, günlük main-lane bütçesi dolmuş kaynağı eligible'dan düşürür.
- Admission control JOB cinsinden (EST_JOBS_PER_VEHICLE=400), araç cinsinden değil.
- Günlük bütçe lane-aware (fast lane'e %20 rezerv) + READ-then-INCR (sayaç artık
  defer'lerle şişmiyor; 48531 vs 30000 gözlemi) + 45dk jitter (thundering herd).
- KÖK NEDEN: addJob artık boolean, queueCategoryJob sayı döndürüyor; progress.total
  yalnız GERÇEKTEN kuyruğa giren job'ı sayıyor. Şişmiş total zinciri asla
  "finished"a ulaştırmıyordu → araç fullyFetched işaretlenmiyor → scan sonsuza dek
  yeniden seçiyordu (~25/gün platosu). processInit queued===0 ise finalizeVehicle.
- prefetch:scheduled:<id> ÇAKIŞMASI: backfill artık :backfill: namespace'i yazıyor;
  eski 6h TTL kullanıcının fresh-decode fast-lane init'ini sessizce atlatıyordu.
  TTL 6h→36h (bütçe-park edilmiş zincir ~24h yaşıyor).
- Phase-2 artık kalıcı vehicles.fullyFetched'i hedefliyor (efemeral marker değil).
- CATEGORY_CAP DB'den ölçülüyor (her processInit'te sıfırlanan sayaçtan değil).

537 test geçti. Bütçe artışı (PREFETCH_DAILY_PCAT) BU PR'da DEĞİL — ayrı adım.

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

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

@@ -17,17 +17,30 @@ function makeDb(limitResults: unknown[][]) {
 }
 
 function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
-  const queue = {
+  // A queue double: getJob returns null (nothing deduped) and the ioredis client
+  // stub answers the delayed-ZSET zcount the pressure gate takes.
+  const makeQueue = (name: string, waiting: number) => ({
+    name,
     // typed args so `.mock.calls[i]` is `unknown[]` (not a 0-length tuple) — the
     // production `nest build` compiles spec files and rejects tuple-index access.
     add: vi.fn((..._args: unknown[]) => Promise.resolve(undefined)),
-    getJobCounts: vi.fn(async () => ({ waiting: opts.waiting, delayed: 0, active: 0 })),
-  };
+    getJob: vi.fn(async (..._args: unknown[]): Promise<unknown> => null),
+    getJobCounts: vi.fn(async (..._args: unknown[]) => ({
+      waiting,
+      delayed: 0,
+      active: 0,
+      prioritized: 0,
+    })),
+    toKey: (t: string) => `bull:${name}:${t}`,
+    client: Promise.resolve({ zcount: vi.fn(async (..._args: unknown[]) => 0) }),
+  });
+  const queue = makeQueue("catalog-prefetch", opts.waiting);
   const redis = {
     // typed args (like queue.add) so mockImplementation((k)=>…) type-checks under
     // the production nest build, which compiles spec files.
     exists: vi.fn(async (..._args: unknown[]) => false), // no cooldown / guard / complete marker
-    get: vi.fn(async () => null), // no no-result residue
+    // null → no no-result residue AND daily budget counters read as 0 (unspent).
+    get: vi.fn(async (..._args: unknown[]): Promise<string | null> => null),
     set: vi.fn(async (..._args: unknown[]) => undefined),
     del: vi.fn(async (..._args: unknown[]) => undefined),
     incr: vi.fn(async (..._args: unknown[]) => 1), // per-source rate window counter
@@ -39,7 +52,7 @@ function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
   };
   const posthog = { payload: vi.fn(async () => ({})) }; // compiled-in defaults
   const db = makeDb(opts.limitResults);
-  const fastQueue = { add: vi.fn(async (..._args: unknown[]) => ({ id: "fastjob" })) };
+  const fastQueue = makeQueue("catalog-prefetch-fast", 0);
   const service = new PrefetchWorkerService(
     queue as never,
     fastQueue as never,
@@ -238,11 +251,10 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
   describe("poison guard (Faz 6: category cap — brand-agnostic anti-explosion)", () => {
     type PC = { processChildren: (j: unknown) => Promise<void> };
 
-    it("processChildren marks poison once progress.total exceeds CATEGORY_CAP", async () => {
-      const { service, queue, redis } = makeDeps({ waiting: 0, limitResults: [] });
-      redis.getJson.mockImplementation(async (...a: unknown[]) =>
-        String(a[0]).includes("progress") ? { total: 5000 } : null,
-      );
+    it("processChildren marks poison once the STORED tree exceeds CATEGORY_CAP", async () => {
+      // The cap is now measured from the DB (categories count), not the ephemeral
+      // progress.total which every processInit resets to 0.
+      const { service, queue, redis } = makeDeps({ waiting: 0, limitResults: [[{ n: 5000 }]] });
       await (service as never as PC).processChildren({
         data: { vehicleId: "op1", categoryId: "c1", source: "pl24", depth: 1 },
       } as never);

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

@@ -53,8 +53,51 @@ const MAX_DEPTH = Number(process.env.PREFETCH_MAX_DEPTH) || 12;
 const BACKFILL_BATCH_SIZE = 40;
 /** Skip the wave entirely if the queue already has more than this many jobs pending. */
 const BACKFILL_MAX_BACKLOG = 1000;
-/** In-flight guard TTL (seconds) — safety net if a run dies without clearing. */
-const BACKFILL_SCHEDULED_TTL = 6 * 60 * 60;
+/**
+ * In-flight guard TTL (seconds) — safety net if a run dies without clearing.
+ *
+ * 36h, NOT 6h: a chain deferred on the daily budget lives until the next UTC
+ * midnight (~24h). With a 6h TTL the guard expired mid-chain, the scan re-picked
+ * the vehicle, processInit reset progress.total, and the in-flight jobs'
+ * incrementCompleted then compared `completed >= total` against the NEW total —
+ * corrupting completion accounting for a still-partial vehicle.
+ */
+const BACKFILL_SCHEDULED_TTL = Number(process.env.PREFETCH_SCHEDULED_TTL_H || 36) * 60 * 60;
+/**
+ * Only delayed jobs due within this horizon count as queue PRESSURE.
+ *
+ * `delayed` lumps two very different things together: per-minute source-rate
+ * defers (<60s) and retry backoff (30/60s), which ARE load, versus daily-budget
+ * and off-hours defers (hours, parked on the next UTC midnight), which are merely
+ * THROTTLED FUTURE WORK. Counting the latter as backlog froze Phase-2 for most of
+ * the day (2026-08-01: delayed=11495, all pcat, all parked at 00:00:01 UTC; the
+ * partial drain fell to ~25 vehicles/day, ETA 79 days).
+ */
+const PRESSURE_HORIZON_MS = Number(process.env.PREFETCH_PRESSURE_HORIZON_MS) || 10 * 60_000;
+/**
+ * Absolute ceiling on TOTAL pending jobs (waiting+active+delayed+prioritized)
+ * across BOTH lanes — the runaway backstop that `delayed` used to provide by
+ * accident. Excluding long defers from the pressure gate removes the only
+ * negative feedback on production, so the pool needs its own hard stop: Phase-2
+ * can otherwise produce ~288k jobs/day against ~65k/day of budget. 50k ≈ 15h of
+ * the combined daily budgets, so the pool always drains inside a window; 1/9 of
+ * the guard-less BFS runaway (470k, months to drain). Set 0 to disable Phase-2.
+ */
+const HARD_MAX_TOTAL_JOBS = Number(process.env.PREFETCH_MAX_TOTAL_JOBS ?? 50_000);
+/**
+ * Rough fan-out of one Phase-2 re-drill. Admission control is done in JOB units,
+ * not vehicle units: 40 vehicles/wave is meaningless when one vehicle is 100-3000
+ * jobs (p95 tree = 744 categories).
+ */
+const EST_JOBS_PER_VEHICLE = Number(process.env.PREFETCH_EST_JOBS_PER_VEHICLE) || 400;
+/**
+ * Share of each source's daily budget reserved for the FAST lane. The daily
+ * counter has no lane component, so backfill spending the budget also parked the
+ * user's fresh-decode chain until midnight. One shared counter (the total
+ * upstream/bandwidth ceiling stays exactly SOURCE_DAILY_MAX) but two thresholds:
+ * the main (backfill) lane stops at 80%, the fast lane may use 100%.
+ */
+const DAILY_FAST_RESERVE = 0.2;
 /** Only these decode sources have catalogs worth prefetching. */
 const BACKFILL_SOURCES = ["pl24", "emex", "parts-catalogs"];
 /** Redis key holding the rolling rescan cursor (last createdAt seen). */
@@ -268,14 +311,14 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           job.name === "prefetch-children" ||
           job.name === "prefetch-parts")
       ) {
-        await this.checkSourceRate(
-          data.source,
-          (job.data as { fast?: boolean }).fast ? "fast" : "main",
-        );
+        const lane = (job.data as { fast?: boolean }).fast ? "fast" : "main";
+        await this.checkSourceRate(data.source, lane);
         // Daily budget AFTER the per-minute gate: a job deferred on the minute
         // ceiling above never reaches here, so rate-limited retries don't inflate
-        // the daily counter — only jobs about to do real work are counted.
-        await this.checkSourceDailyBudget(data.source);
+        // the daily counter — only jobs about to do real work are counted. The
+        // lane decides which threshold applies (backfill stops at the main limit,
+        // the user's fast lane may use the full budget).
+        await this.checkSourceDailyBudget(data.source, lane);
       }
       if (data.source === "parts-catalogs" && PCAT_PACE_MS > 0) {
         await new Promise((r) => setTimeout(r, PCAT_PACE_MS));
@@ -407,8 +450,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
         for (const child of children) {
           if (child.unavailable) continue;
-          await this.queueCategoryJob(child, vehicleId, source, 1, fast);
-          queued++;
+          // Count jobs ACTUALLY queued, not nodes walked — see addJob's doc: an
+          // inflated progress.total makes the chain never reach "finished".
+          queued += await this.queueCategoryJob(child, vehicleId, source, 1, fast);
         }
       } else if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups)) {
         // Leaf — check if parts already fetched
@@ -418,32 +462,46 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           .where(eq(parts.categoryId, cat.id))
           .limit(1);
 
-        if (!partCheck && cat.linkPath) {
-          await this.addJob("prefetch-parts", {
+        if (
+          !partCheck &&
+          cat.linkPath &&
+          (await this.addJob("prefetch-parts", {
             vehicleId,
             categoryId: cat.id,
             source,
             action: "parts" as const,
             depth: 0,
             fast,
-          });
+          }))
+        ) {
           queued++;
         }
-      } else if (cat.linkPath) {
+      } else if (
+        cat.linkPath &&
         // Non-leaf without children — needs children fetch
-        await this.addJob("prefetch-children", {
+        (await this.addJob("prefetch-children", {
           vehicleId,
           categoryId: cat.id,
           source,
           action: "children" as const,
           depth: 0,
           fast,
-        });
+        }))
+      ) {
         queued++;
       }
     }
 
     await updateProgress(this.redis, vehicleId, { total: queued });
+    if (queued === 0) {
+      // Nothing left to fetch — the tree is already complete in DB. Without this
+      // there is no job to fire incrementCompleted, so progress sits at {0,0}, the
+      // vehicle is never marked fullyFetched, and the scan re-picks it every wave
+      // forever while burning a daily-budget slot each time.
+      this.logger.log(`[prefetch] Nothing to queue for vehicle=${vehicleId} — already complete`);
+      await this.finalizeVehicle(vehicleId);
+      return;
+    }
     this.logger.log(`[prefetch] Queued ${queued} sub-jobs for vehicle=${vehicleId}`);
   }
 
@@ -462,14 +520,19 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    // Anti-poison cap: progress.total tracks every sub-job queued for this vehicle
-    // (≈ its discovered tree size). Once it blows past the ceiling the vehicle is a
-    // generic-catalog explosion — stop drilling and mark it poison so the rest of
-    // its (part-less) tree is never fetched and it's never re-picked.
-    const prog = await this.redis.getJson<{ total: number }>(`prefetch:progress:${vehicleId}`);
-    if ((prog?.total ?? 0) >= CATEGORY_CAP) {
+    // Anti-poison cap measured from the DB, NOT from progress.total: the ephemeral
+    // counter is reset to 0 by every processInit, so a re-picked vehicle could
+    // drill another CATEGORY_CAP nodes per round and never trip the guard (the
+    // generic-ROOT Opels: 420k/102k/21k categories, ZERO parts). The stored tree
+    // is the real, cumulative size.
+    const [capRow] = await this.db
+      .select({ n: sql<number>`count(*)::int` })
+      .from(categories)
+      .where(eq(categories.vehicleId, vehicleId))
+      .limit(1);
+    if ((capRow?.n ?? 0) >= CATEGORY_CAP) {
       this.logger.warn(
-        `[prefetch] Category cap ${CATEGORY_CAP} hit for vehicle=${vehicleId} (tree≈${prog?.total}) — marking poison, stop drilling`,
+        `[prefetch] Category cap ${CATEGORY_CAP} hit for vehicle=${vehicleId} (tree=${capRow?.n}) — marking poison, stop drilling`,
       );
       await this.markPoison(vehicleId);
       return;
@@ -481,8 +544,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       let queued = 0;
       for (const child of children) {
         if (child.unavailable) continue;
-        await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
-        queued++;
+        // Real queued-job count (see addJob) — walking a node that dedupes or
+        // already has parts must not inflate progress.total.
+        queued += await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
       }
 
       if (queued > 0) {
@@ -561,19 +625,57 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     // genuinely-empty vehicles keep getting onboarded through the fast lane even
     // while a large deep-drill backlog is still draining — otherwise a single
     // backlog spike freezes new-vehicle coverage until the whole queue clears.
-    const counts = await this.queue.getJobCounts("waiting", "delayed", "active");
-    const backlog = (counts.waiting ?? 0) + (counts.delayed ?? 0) + (counts.active ?? 0);
-    const phase2Allowed = backlog <= maxBacklog;
+    // Two INDEPENDENT ceilings (Phase-1 fast lane is never gated, so genuinely
+    // empty vehicles keep getting onboarded):
+    //  1. pressure = waiting + active + delayed-due-within-PRESSURE_HORIZON_MS,
+    //     both lanes, vs maxBacklog — "are the workers actually swamped?".
+    //     Budget/off-hours defers are parked hours out and no longer count, so a
+    //     spent daily budget stops freezing Phase-2 for the rest of the day.
+    //  2. total = every pending job, both lanes, vs HARD_MAX_TOTAL_JOBS — the
+    //     runaway backstop that `delayed` used to provide by accident.
+    const depth = await this.getQueueDepth();
+    const headroom = Math.max(0, HARD_MAX_TOTAL_JOBS - depth.total);
+    // Admission control in JOB units: one wave fans out to batchSize * ~400 jobs
+    // long before the next hourly scan can react, so shrink the wave as the pool
+    // fills instead of stepping over the cap by a whole batch.
+    const phase2Budget = Math.min(batchSize, Math.floor(headroom / EST_JOBS_PER_VEHICLE));
+    const phase2Allowed = depth.pressure <= maxBacklog && phase2Budget > 0;
     if (!phase2Allowed) {
-      this.logger.log(`[backfill] Backlog ${backlog} > ${maxBacklog} — Phase-1 (fast lane) only`);
+      this.logger.log(
+        `[backfill] Phase-1 (fast lane) only — pressure=${depth.pressure}/${maxBacklog}, ` +
+          `total=${depth.total}/${HARD_MAX_TOTAL_JOBS} (delayed=${depth.delayed}, imminent=${depth.imminent})`,
+      );
+    }
+    if (HARD_MAX_TOTAL_JOBS > 0 && depth.total > HARD_MAX_TOTAL_JOBS * 1.5) {
+      // Should be unreachable — admission control caps intake once per hour. If it
+      // fires, real fan-out is far above EST_JOBS_PER_VEHICLE.
+      this.logger.error(
+        `[backfill] Queue pool ${depth.total} > 1.5x hard cap ${HARD_MAX_TOTAL_JOBS} — investigate fan-out`,
+      );
     }
 
-    // Only target sources eligible right now: not in cooldown (user active) and
-    // inside their scrape window (PL24/parts-catalogs office hours; EMEX always).
+    // Only target sources eligible right now: not in cooldown (user active),
+    // inside their scrape window, AND still inside today's main-lane budget.
+    // The budget check is THE key guard: deferring at job level only MOVES work
+    // into `delayed`, it does not stop PRODUCING it — and budget defers never
+    // exhaust attempts (moveToDelayed skipAttempt), so nothing drops them. Once a
+    // source's main-lane budget is spent the scan must stop feeding it for the
+    // rest of the UTC day; feeding a throttled source is exactly how the 470k
+    // runaway was built. This also preserves the fast-lane reserve for real users.
     const eligible: string[] = [];
     for (const s of BACKFILL_SOURCES) {
       if (await this.redis.exists(`prefetch:activity:${s}`)) continue;
       if (cfg.businessHoursOnly !== false && !isWithinTimeWindow(s)) continue;
+      const mainLimit = this.dailyMainLimit(s);
+      if (mainLimit > 0) {
+        const spent = Number((await this.redis.get(this.dailyKey(s))) ?? 0);
+        if (spent >= mainLimit) {
+          this.logger.log(
+            `[backfill] ${s} daily budget spent (${spent}/${mainLimit}) — source skipped this wave`,
+          );
+          continue;
+        }
+      }
       eligible.push(s);
     }
     if (eligible.length === 0) {
@@ -588,9 +690,17 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     const tryPick = async (
       v: { id: string; source: string | null },
       fast: boolean,
+      limit: number = batchSize,
     ): Promise<void> => {
-      if (picked.length >= batchSize || seen.has(v.id) || !v.source) return;
-      if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // already in flight
+      if (picked.length >= limit || seen.has(v.id) || !v.source) return;
+      // TWO different in-flight guards, both must be clear:
+      //  - prefetch:scheduled:backfill:<id> — ours (36h, covers a budget-parked chain)
+      //  - prefetch:scheduled:<id>          — the USER decode path's (vehicles.service.ts
+      //    schedulePrefetch, short TTL). We deliberately stopped WRITING the bare
+      //    key: our long TTL made schedulePrefetch() silently skip the user's
+      //    fresh-decode fast-lane init.
+      if (await this.redis.exists(this.scheduledKey(v.id))) return;
+      if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // user chain in flight
       // Skip exhausted residue: vehicles whose prefetch keeps finishing with zero
       // parts (no catalog data). They'd otherwise be re-picked every wave forever.
       const noResult = await this.redis.get(this.noResultKey(v.id));
@@ -623,17 +733,27 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
     for (const v of noParts) await tryPick(v, true);
 
-    // Phase 2 — rolling rescan of ALL decoded vehicles to gap-fill partially-fetched
-    // ones. A createdAt cursor walks forward and wraps around at the end. Gated by
-    // the backlog ceiling (above) so it doesn't pile on while the queue is deep.
-    if (phase2Allowed && picked.length < batchSize) {
+    // Phase 2 — rolling rescan of vehicles that are NOT fully fetched, to gap-fill
+    // partials. Targets the DURABLE `fullyFetched` flag (indexed) instead of
+    // walking the whole fleet and relying only on the ephemeral 21-day
+    // `prefetch:complete:` marker — a Redis flush would otherwise re-drill every
+    // vehicle at once. The fullyFetchedAt clause keeps the periodic re-validation
+    // the TTL used to provide. Gated by the ceilings above + a JOB-unit budget.
+    const phase2Limit = Math.min(batchSize, picked.length + phase2Budget);
+    if (phase2Allowed && picked.length < phase2Limit) {
       const cursorObj = await this.redis.getJson<{ ts: string }>(BACKFILL_CURSOR_KEY);
       const cursor = cursorObj?.ts ? new Date(cursorObj.ts) : new Date(0);
 
       const rolling = await this.db
         .select({ id: vehicles.id, source: vehicles.source, createdAt: vehicles.createdAt })
         .from(vehicles)
-        .where(and(inArray(vehicles.source, eligible), gt(vehicles.createdAt, cursor)))
+        .where(
+          and(
+            inArray(vehicles.source, eligible),
+            gt(vehicles.createdAt, cursor),
+            sql`(${vehicles.fullyFetched} = false OR ${vehicles.fullyFetchedAt} < now() - interval '21 days')`,
+          ),
+        )
         .orderBy(asc(vehicles.createdAt))
         .limit(overfetch);
 
@@ -649,7 +769,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       let lastTs: Date | null = null;
       for (const v of rolling) {
         lastTs = v.createdAt;
-        await tryPick(v, false);
+        await tryPick(v, false, phase2Limit);
       }
       if (lastTs) {
         await this.redis.setJson(BACKFILL_CURSOR_KEY, { ts: lastTs.toISOString() }, 30 * 86400);
@@ -665,7 +785,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     const fastCount = picked.filter((v) => v.fast).length;
     this.logger.log(
       `[backfill] Queued ${picked.length} vehicle(s) (${fastCount} fast-lane, ` +
-        `sources=${eligible.join(",")}, backlog=${backlog})`,
+        `sources=${eligible.join(",")}, pressure=${depth.pressure}, total=${depth.total}, ` +
+        `phase2Budget=${phase2Budget})`,
     );
   }
 
@@ -684,7 +805,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       },
     );
     // Guard cleared on completion (incrementCompleted) or by TTL if the run dies.
-    await this.redis.set(`prefetch:scheduled:${vehicleId}`, "1", BACKFILL_SCHEDULED_TTL);
+    await this.redis.set(this.scheduledKey(vehicleId), "1", BACKFILL_SCHEDULED_TTL);
   }
 
   // ==================== Helpers ====================
@@ -701,8 +822,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     source: string,
     depth: number,
     fast = false,
-  ): Promise<void> {
-    if (cat.unavailable) return;
+  ): Promise<number> {
+    if (cat.unavailable) return 0;
 
     if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups)) {
       // Leaf — check if already has parts
@@ -713,16 +834,20 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
         .limit(1);
 
       if (!partCheck && cat.linkPath) {
-        await this.addJob("prefetch-parts", {
+        return (await this.addJob("prefetch-parts", {
           vehicleId,
           categoryId: cat.id,
           source: source as "pl24" | "emex",
           action: "parts" as const,
           depth,
           fast,
-        });
+        }))
+          ? 1
+          : 0;
       }
-    } else if (cat.linkPath && depth < MAX_DEPTH) {
+      return 0;
+    }
+    if (cat.linkPath && depth < MAX_DEPTH) {
       // Non-leaf within the depth cap — explore children. The `depth < MAX_DEPTH`
       // gate mirrors processChildren's early-return: without it we'd enqueue a
       // prefetch-children job that processChildren just drops, burning a rate-limit
@@ -740,21 +865,25 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           .from(categories)
           .where(eq(categories.parentId, cat.id));
 
+        let n = 0;
         for (const child of children) {
           if (child.unavailable) continue;
-          await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
+          n += await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
         }
-      } else {
-        await this.addJob("prefetch-children", {
-          vehicleId,
-          categoryId: cat.id,
-          source: source as "pl24" | "emex",
-          action: "children" as const,
-          depth,
-          fast,
-        });
+        return n;
       }
+      return (await this.addJob("prefetch-children", {
+        vehicleId,
+        categoryId: cat.id,
+        source: source as "pl24" | "emex",
+        action: "children" as const,
+        depth,
+        fast,
+      }))
+        ? 1
+        : 0;
     }
+    return 0;
   }
 
   private isLeafLinkPath(
@@ -785,12 +914,25 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     );
   }
 
-  private async addJob(name: string, data: PrefetchCategoryJobData): Promise<void> {
+  /**
+   * Queue one sub-job. Returns whether a NEW job was actually created.
+   *
+   * The return value is the fix for a completion-accounting bug: `progress.total`
+   * is the chain's denominator (incrementCompleted fires "finished" at
+   * `completed >= total`), but callers used to increment it for every node they
+   * WALKED — including nodes where this deterministic jobId deduped the add, and
+   * leaves that already had parts. An inflated total means the chain never
+   * reaches "finished", so the vehicle is never marked fullyFetched and the scan
+   * re-picks it every wave forever (the ~25 vehicles/day plateau). Queue
+   * behaviour is UNCHANGED — BullMQ already no-op'd a duplicate jobId.
+   */
+  private async addJob(name: string, data: PrefetchCategoryJobData): Promise<boolean> {
+    // BullMQ rejects custom job IDs containing ":" (its key separator), so use
+    // "-" instead. The values are UUIDs — the ID only needs to be deterministic
+    // (for dedup), not parseable.
+    const jobId = `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`;
     const opts: Record<string, unknown> = {
-      // BullMQ rejects custom job IDs containing ":" (its key separator), so use
-      // "-" instead. The values are UUIDs — the ID only needs to be deterministic
-      // (for dedup), not parseable.
-      jobId: `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`,
+      jobId,
       // Fast lane (Phase-1 / reactive): add with `lifo` so the job RPUSHes to the
       // TAIL of the wait list, where BullMQ's RPOPLPUSH picks it next — i.e. ahead
       // of the deep deep-drill backlog already sitting in wait. (BullMQ 5 drains
@@ -798,16 +940,15 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       // OPPOSITE here and starve the job behind the backlog; lifo is correct.)
       ...(data.fast ? { lifo: true } : {}),
     };
-    if (data.fast) {
-      await this.fastQueue.add(name, data, opts);
-      return;
-    }
+    const q = data.fast ? this.fastQueue : this.queue;
+    if (await q.getJob(jobId)) return false;
 
     // parts-catalogs pacing is handled per-job in process() (PCAT_PACE_MS) + the
     // limiter. The old cumulative `index * 20s` delay was pathological (the Nth
     // leaf of a vehicle waited N*20s) and is gone.
 
-    await this.queue.add(name, data, opts);
+    await q.add(name, data, opts);
+    return true;
   }
 
   private async incrementCompleted(vehicleId: string): Promise<void> {
@@ -827,36 +968,47 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
     if (isFinished) {
       this.logger.log(`[prefetch] Completed all jobs for vehicle=${vehicleId}`);
-      // Genuine residue: the whole chain finished but the vehicle still has no
-      // parts (all leaves empty / no catalog data). Count it so Phase-1 stops
-      // re-picking it every wave.
-      const [hasPart] = await this.db
-        .select({ id: parts.id })
-        .from(parts)
-        .where(eq(parts.vehicleId, vehicleId))
-        .limit(1);
-      if (hasPart) {
-        // Fully fetched with parts → mark complete so the Phase-2 rescan skips it
-        // (re-validates after the TTL). A chain with any failed job never reaches
-        // isFinished, so partially-fetched vehicles are never marked — they keep
-        // getting gap-filled.
-        await this.redis.set(this.completeKey(vehicleId), "1", COMPLETE_TTL_S);
-        // Durable completeness flag (never expires) — the reliable "% fully
-        // fetched" measurement, independent of the ephemeral Redis marker.
-        // Re-set on every re-drill completion so fullyFetchedAt tracks last-verified.
-        await this.db
-          .update(vehicles)
-          .set({ fullyFetched: true, fullyFetchedAt: new Date() })
-          .where(eq(vehicles.id, vehicleId));
-      } else {
-        await this.markNoResult(vehicleId);
-      }
-      // Clean up Redis keys — data is in PostgreSQL now
-      await this.redis.del(`prefetch:scheduled:${vehicleId}`);
-      await this.redis.del(`prefetch:progress:${vehicleId}`);
+      await this.finalizeVehicle(vehicleId);
     }
   }
 
+  /**
+   * Settle a vehicle whose prefetch chain is done: mark it complete (with parts)
+   * or count it as residue (still zero parts), then clear the run's Redis state.
+   * Called both when the last sub-job finishes AND when processInit finds there
+   * is nothing left to queue — otherwise an already-complete tree would never be
+   * settled and the scan would re-pick it forever.
+   */
+  private async finalizeVehicle(vehicleId: string): Promise<void> {
+    // Genuine residue: the whole chain finished but the vehicle still has no
+    // parts (all leaves empty / no catalog data). Count it so Phase-1 stops
+    // re-picking it every wave.
+    const [hasPart] = await this.db
+      .select({ id: parts.id })
+      .from(parts)
+      .where(eq(parts.vehicleId, vehicleId))
+      .limit(1);
+    if (hasPart) {
+      // Fully fetched with parts → mark complete so the Phase-2 rescan skips it
+      // (re-validates after the TTL). A chain with any failed job never reaches
+      // isFinished, so partially-fetched vehicles are never marked — they keep
+      // getting gap-filled.
+      await this.redis.set(this.completeKey(vehicleId), "1", COMPLETE_TTL_S);
+      // Durable completeness flag (never expires) — the reliable "% fully
+      // fetched" measurement, independent of the ephemeral Redis marker.
+      // Re-set on every re-drill completion so fullyFetchedAt tracks last-verified.
+      await this.db
+        .update(vehicles)
+        .set({ fullyFetched: true, fullyFetchedAt: new Date() })
+        .where(eq(vehicles.id, vehicleId));
+    } else {
+      await this.markNoResult(vehicleId);
+    }
+    // Clean up Redis keys — data is in PostgreSQL now
+    await this.redis.del(this.scheduledKey(vehicleId));
+    await this.redis.del(`prefetch:progress:${vehicleId}`);
+  }
+
   private async incrementErrors(vehicleId: string): Promise<void> {
     const progress = await this.redis.getJson<{ errors: number }>(`prefetch:progress:${vehicleId}`);
     if (!progress) return;
@@ -869,6 +1021,62 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     return `prefetch:noresult:${vehicleId}`;
   }
 
+  /**
+   * Backfill's own in-flight guard — namespaced so it can't shadow the user
+   * decode path's `prefetch:scheduled:<id>` (vehicles.service.ts reads that key
+   * and skips the fresh-decode fast-lane init while it is set).
+   */
+  private scheduledKey(vehicleId: string): string {
+    return `prefetch:scheduled:backfill:${vehicleId}`;
+  }
+
+  /**
+   * Queue depth across BOTH lanes, split into "pressure" (work the workers will
+   * pick up within PRESSURE_HORIZON_MS) and "total" (everything pending).
+   *
+   * BullMQ stores delayed jobs in a ZSET scored `dueMs * 0x1000 + seq` (12-bit
+   * collision counter — addDelayedJob / moveToDelayed lua), so "due within X ms"
+   * is a plain ZCOUNT with the bound encoded the same way. RedisService exposes
+   * no zcount, so we borrow each queue's OWN ioredis connection and its toKey()
+   * rather than hand-building `bull:<name>:delayed`.
+   */
+  private async getQueueDepth(): Promise<{
+    pressure: number;
+    total: number;
+    imminent: number;
+    delayed: number;
+  }> {
+    // String bound so ioredis can't render it in exponent notation.
+    const maxScore = String((Date.now() + PRESSURE_HORIZON_MS + 1) * 0x1000 - 1);
+    let pressure = 0;
+    let total = 0;
+    let imminent = 0;
+    let delayed = 0;
+    for (const q of [this.queue, this.fastQueue]) {
+      const c = await q.getJobCounts("waiting", "active", "delayed", "prioritized");
+      const live = (c.waiting ?? 0) + (c.active ?? 0);
+      const d = c.delayed ?? 0;
+      // Fail CLOSED: if the ZCOUNT can't be taken, count every delayed job as
+      // pressure — fall back to the old over-conservative gate rather than
+      // silently unlocking Phase-2 with no visibility. A repeating warn here
+      // means Phase-2 is suspended again.
+      let due = d;
+      try {
+        const client = await q.client;
+        due = await client.zcount(q.toKey("delayed"), "-inf", maxScore);
+      } catch (err) {
+        this.logger.warn(
+          `[backfill] delayed zcount failed on ${q.name} (${(err as Error).message}) — counting all delayed as pressure`,
+        );
+      }
+      pressure += live + due;
+      total += live + d + (c.prioritized ?? 0);
+      imminent += due;
+      delayed += d;
+    }
+    return { pressure, total, imminent, delayed };
+  }
+
   private completeKey(vehicleId: string): string {
     return `prefetch:complete:${vehicleId}`;
   }
@@ -884,7 +1092,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
    *  (which stay ~200 categories and DO yield parts) are never affected. */
   private async markPoison(vehicleId: string): Promise<void> {
     await this.redis.set(this.poisonKey(vehicleId), "1", POISON_TTL_S);
-    await this.redis.del(`prefetch:scheduled:${vehicleId}`);
+    await this.redis.del(this.scheduledKey(vehicleId));
     await this.redis.del(`prefetch:progress:${vehicleId}`);
   }
 
@@ -918,26 +1126,52 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
    * when the source's daily job count exceeds its ceiling, defer the job until
    * the window rolls so a long run can't drain the proxy budget. 0 = unlimited.
    */
-  private async checkSourceDailyBudget(source: string): Promise<void> {
+  private async checkSourceDailyBudget(
+    source: string,
+    lane: "main" | "fast" = "main",
+  ): Promise<void> {
     const max = SOURCE_DAILY_MAX[source] ?? 0;
     if (max <= 0) return;
+    const limit = lane === "fast" ? max : this.dailyMainLimit(source);
     const dayMs = 86_400_000;
     const now = Date.now();
-    const key = `prefetch:daily:${source}:${Math.floor(now / dayMs)}`;
-    const n = await this.redis.incr(key);
-    if (n === 1) await this.redis.expire(key, 90_000); // ~25h, outlives the window
-    if (n > max) {
-      // Defer until the next UTC day (+1s slack); other sources keep flowing.
-      const msLeft = dayMs - (now % dayMs) + 1000;
-      if (n === max + 1) {
+    const key = this.dailyKey(source, now);
+
+    // READ-then-INCR (was INCR-then-check). A REJECTED attempt must not count:
+    // the old order inflated the counter with every defer (observed 48531 against
+    // a 30000 budget), which (a) made the number useless for capacity decisions
+    // and (b) — now that the scan reads the same counter to stop feeding a spent
+    // source — would let pure defer churn lock the source out. Worst-case
+    // overshoot under the read/incr race is WORKER_CONCURRENCY jobs: acceptable.
+    const n = Number((await this.redis.get(key)) ?? 0);
+    if (n >= limit) {
+      // Defer to the next UTC day, plus up to 45min of JITTER. Without jitter every
+      // deferred job wakes in the SAME millisecond (observed: 11495 jobs all at
+      // 00:00:01 UTC) — the promotion lands as one burst and the pressure signal
+      // flaps. Other sources keep flowing (per-job defer, not a worker pause).
+      const msLeft = dayMs - (now % dayMs) + 1000 + Math.floor(Math.random() * 45 * 60_000);
+      if (n === limit) {
         this.logger.warn(
-          `[prefetch] ${source} hit daily budget (${max}) — deferring backfill ~${Math.round(
-            msLeft / 3_600_000,
-          )}h until window rolls`,
+          `[prefetch] ${source} daily budget hit (lane=${lane}, ${n}/${limit} of ${max}) — ` +
+            `deferring ~${Math.round(msLeft / 3_600_000)}h until the window rolls`,
         );
       }
       throw new RateLimitError(msLeft, "source-rate");
     }
+    const after = await this.redis.incr(key);
+    if (after === 1) await this.redis.expire(key, 90_000); // ~25h, outlives the window
+  }
+
+  /** Redis key for a source's UTC-day budget counter (shared by both lanes). */
+  private dailyKey(source: string, now = Date.now()): string {
+    return `prefetch:daily:${source}:${Math.floor(now / 86_400_000)}`;
+  }
+
+  /** Main (backfill) lane threshold — the fast lane's reserve is never available
+   *  to backfill, so a sweep can't park the user's fresh-decode chain. */
+  private dailyMainLimit(source: string): number {
+    const max = SOURCE_DAILY_MAX[source] ?? 0;
+    return max <= 0 ? 0 : Math.floor(max * (1 - DAILY_FAST_RESERVE));
   }
 
   /** Record that a backfill attempt finished with the vehicle still at zero parts. */