perf(backfill): fast-lane Phase-1 via lifo + run scan despite deep backlog

The hourly catalog-backfill scan had gone effectively dead on prod: its
job sat behind a ~430k-deep wait list and, even when it ran, self-skipped
because the queue backlog (431k) was far over the maxBacklog ceiling
(1000). Net result: newly-decoded / zero-parts vehicles were never
onboarded — they starved behind the deep-drill backlog (ic stuck at 28
for days, 0 backfill-scan jobs ever completed).

Root constraint (verified against bullmq 5.68 lua): moveToActive drains
the `wait` list (RPOPLPUSH from the tail) BEFORE the `prioritized` ZSET,
so a `priority` job is starved behind an already-deep wait queue — the
opposite of what's wanted. The lever that works is `lifo`: it RPUSHes to
the tail, where the very next RPOPLPUSH picks it, ahead of the FIFO
backlog. addJobFromScheduler honours lifo too, so the scan job itself can
jump the queue.

Changes:
- Thread a `fast` flag through the init→children→parts chain; fast jobs
  are enqueued with `lifo:true` so the whole chain jumps the backlog.
- Backfill scan: Phase-1 (zero-parts vehicles) now runs every wave in the
  fast lane even when the backlog is over the ceiling; only Phase-2 (the
  rolling rescan that piles on) is suspended while the queue is deep.
- Register the hourly scan job with `lifo:true` so it fires on the next
  tick instead of being buried for days.

Tested: new prefetch-worker.service.spec (lifo wiring + Phase-1/Phase-2
gating), full api suite green (248 passed), typecheck + biome clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -167,7 +167,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
    * Init job: walk the category tree for a vehicle and queue sub-jobs.
    */
   private async processInit(job: Job<PrefetchInitJobData>): Promise<void> {
-    const { vehicleId, source } = job.data;
+    const { vehicleId, source, fast = false } = job.data;
     this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`);
 
     await checkCooldown(this.redis, source);
@@ -244,7 +244,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
         for (const child of children) {
           if (child.unavailable) continue;
-          await this.queueCategoryJob(child, vehicleId, source, 1);
+          await this.queueCategoryJob(child, vehicleId, source, 1, fast);
           queued++;
         }
       } else if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups)) {
@@ -262,6 +262,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
             source,
             action: "parts" as const,
             depth: 0,
+            fast,
           });
           queued++;
         }
@@ -273,6 +274,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           source,
           action: "children" as const,
           depth: 0,
+          fast,
         });
         queued++;
       }
@@ -286,7 +288,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
    * Fetch children (sub-categories) for a category.
    */
   private async processChildren(job: Job<PrefetchCategoryJobData>): Promise<void> {
-    const { vehicleId, categoryId, source, depth } = job.data;
+    const { vehicleId, categoryId, source, depth, fast = false } = job.data;
     this.logger.log(`[prefetch] Children for category=${categoryId}, depth=${depth}`);
 
     await checkCooldown(this.redis, source);
@@ -303,7 +305,7 @@ 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);
+        await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
         queued++;
       }
 
@@ -378,12 +380,16 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
         ? cfg.maxBacklog
         : BACKFILL_MAX_BACKLOG;
 
-    // Self-throttle: don't pile on if the queue is already deep — let it drain.
+    // Self-throttle: when the queue is already deep, suspend only the Phase-2
+    // rolling rescan (the part that piles on). Phase-1 still runs every wave so
+    // 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);
-    if (backlog > maxBacklog) {
-      this.logger.log(`[backfill] Skip — queue backlog ${backlog} > ${maxBacklog}`);
-      return;
+    const phase2Allowed = backlog <= maxBacklog;
+    if (!phase2Allowed) {
+      this.logger.log(`[backfill] Backlog ${backlog} > ${maxBacklog} — Phase-1 (fast lane) only`);
     }
 
     // Only target sources eligible right now: not in cooldown (user active) and
@@ -399,11 +405,14 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    const picked: Array<{ id: string; source: string }> = [];
+    const picked: Array<{ id: string; source: string; fast: boolean }> = [];
     const seen = new Set<string>();
     const overfetch = batchSize * 4; // headroom for in-flight skips
 
-    const tryPick = async (v: { id: string; source: string | null }): Promise<void> => {
+    const tryPick = async (
+      v: { id: string; source: string | null },
+      fast: boolean,
+    ): 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
       // Skip exhausted residue: vehicles whose prefetch keeps finishing with zero
@@ -411,7 +420,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       const noResult = await this.redis.get(this.noResultKey(v.id));
       if (noResult && Number(noResult) >= NORESULT_MAX_ATTEMPTS) return;
       seen.add(v.id);
-      picked.push({ id: v.id, source: v.source });
+      picked.push({ id: v.id, source: v.source, fast });
     };
 
     // Phase 1 — clear the obvious backlog first: decoded vehicles with zero parts.
@@ -429,11 +438,12 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       .orderBy(asc(vehicles.createdAt))
       .limit(overfetch);
 
-    for (const v of noParts) await tryPick(v);
+    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.
-    if (picked.length < batchSize) {
+    // 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) {
       const cursorObj = await this.redis.getJson<{ ts: string }>(BACKFILL_CURSOR_KEY);
       const cursor = cursorObj?.ts ? new Date(cursorObj.ts) : new Date(0);
 
@@ -456,7 +466,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       let lastTs: Date | null = null;
       for (const v of rolling) {
         lastTs = v.createdAt;
-        await tryPick(v);
+        await tryPick(v, false);
       }
       if (lastTs) {
         await this.redis.setJson(BACKFILL_CURSOR_KEY, { ts: lastTs.toISOString() }, 30 * 86400);
@@ -468,18 +478,26 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    for (const v of picked) await this.enqueueInit(v.id, v.source);
+    for (const v of picked) await this.enqueueInit(v.id, v.source, v.fast);
+    const fastCount = picked.filter((v) => v.fast).length;
     this.logger.log(
-      `[backfill] Queued ${picked.length} vehicle(s) (sources=${eligible.join(",")}, backlog=${backlog})`,
+      `[backfill] Queued ${picked.length} vehicle(s) (${fastCount} fast-lane, ` +
+        `sources=${eligible.join(",")}, backlog=${backlog})`,
     );
   }
 
   /** Queue a prefetch-init for a vehicle and set the in-flight guard. */
-  private async enqueueInit(vehicleId: string, source: string): Promise<void> {
+  private async enqueueInit(vehicleId: string, source: string, fast = false): Promise<void> {
     await this.queue.add(
       "prefetch-init",
-      { vehicleId, source: source as PrefetchInitJobData["source"] },
-      { removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 } },
+      { vehicleId, source: source as PrefetchInitJobData["source"], fast },
+      {
+        removeOnComplete: { count: 1000 },
+        removeOnFail: { count: 5000 },
+        // Fast lane (Phase-1): lifo so a zero-parts vehicle's whole chain jumps
+        // the deep-drill backlog instead of queueing behind it (see addJob).
+        ...(fast ? { lifo: true } : {}),
+      },
     );
     // Guard cleared on completion (incrementCompleted) or by TTL if the run dies.
     await this.redis.set(`prefetch:scheduled:${vehicleId}`, "1", BACKFILL_SCHEDULED_TTL);
@@ -498,6 +516,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     vehicleId: string,
     source: string,
     depth: number,
+    fast = false,
   ): Promise<void> {
     if (cat.unavailable) return;
 
@@ -516,6 +535,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           source: source as "pl24" | "emex",
           action: "parts" as const,
           depth,
+          fast,
         });
       }
     } else if (cat.linkPath) {
@@ -535,7 +555,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
         for (const child of children) {
           if (child.unavailable) continue;
-          await this.queueCategoryJob(child, vehicleId, source, depth + 1);
+          await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
         }
       } else {
         await this.addJob("prefetch-children", {
@@ -544,6 +564,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           source: source as "pl24" | "emex",
           action: "children" as const,
           depth,
+          fast,
         });
       }
     }
@@ -583,6 +604,12 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       // "-" instead. The values are UUIDs — the ID only needs to be deterministic
       // (for dedup), not parseable.
       jobId: `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`,
+      // 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
+      // the wait list before the prioritized ZSET, so `priority` would do the
+      // OPPOSITE here and starve the job behind the backlog; lifo is correct.)
+      ...(data.fast ? { lifo: true } : {}),
     };
 
     // parts-catalogs pacing is handled per-job in process() (PCAT_PACE_MS) + the