Merge pull request 'perf(backfill): fast-lane Phase-1 via lifo + run scan despite deep backlog' (#133) from fix/backfill-lifo-fast-lane into main

Reviewed-on: #133

apps/api/src/jobs/jobs.module.ts

@@ -119,6 +119,12 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
           opts: {
             removeOnComplete: { count: 48 },
             removeOnFail: { count: 100 },
+            // lifo so the scan job lands at the tail of the wait list and runs on
+            // the next tick instead of queueing behind the deep-drill backlog —
+            // otherwise the hourly scan is buried for days and never fires. BullMQ
+            // drains wait (RPOPLPUSH from the tail) before the prioritized ZSET, so
+            // lifo — not `priority` — is what jumps an already-deep wait queue.
+            lifo: true,
           },
         },
       );

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

@@ -0,0 +1,141 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { PrefetchWorkerService } from "./prefetch-worker.service";
+
+/**
+ * Chainable drizzle mock: select/from/where/orderBy return the chain; the
+ * terminal limit() yields the next queued result array. Mirrors the trick the
+ * other api specs use (await on a plain array resolves to the array itself).
+ */
+function makeDb(limitResults: unknown[][]) {
+  const queued = [...limitResults];
+  const chain: Record<string, unknown> = {};
+  for (const m of ["select", "from", "where", "orderBy"]) {
+    chain[m] = vi.fn(() => chain);
+  }
+  chain.limit = vi.fn(() => queued.shift() ?? []);
+  return chain;
+}
+
+function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
+  const queue = {
+    add: vi.fn(async () => undefined),
+    getJobCounts: vi.fn(async () => ({ waiting: opts.waiting, delayed: 0, active: 0 })),
+  };
+  const redis = {
+    exists: vi.fn(async () => false), // no source cooldown, no in-flight guard
+    get: vi.fn(async () => null), // no no-result residue
+    set: vi.fn(async () => undefined),
+    getJson: vi.fn(async () => null), // Phase-2 cursor empty
+    setJson: vi.fn(async () => undefined),
+  };
+  const posthog = { payload: vi.fn(async () => ({})) }; // compiled-in defaults
+  const db = makeDb(opts.limitResults);
+  const service = new PrefetchWorkerService(
+    queue as never,
+    {} as never, // categoriesService — unused by the scan
+    redis as never,
+    posthog as never,
+    db as never,
+  );
+  return { service, queue, redis, posthog, db };
+}
+
+describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    process.env.CATALOG_BACKFILL_ENABLED = "true"; // force the prod-host gate on
+  });
+  afterEach(() => {
+    process.env.CATALOG_BACKFILL_ENABLED = undefined;
+  });
+
+  describe("enqueueInit", () => {
+    it("sets lifo + data.fast when fast", async () => {
+      const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
+      await (service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> })
+        .enqueueInit("v1", "emex", true);
+      const [name, data, jobOpts] = queue.add.mock.calls[0];
+      expect(name).toBe("prefetch-init");
+      expect(data).toMatchObject({ vehicleId: "v1", source: "emex", fast: true });
+      expect(jobOpts).toMatchObject({ lifo: true });
+    });
+
+    it("omits lifo when not fast", async () => {
+      const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
+      await (service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> })
+        .enqueueInit("v1", "emex", false);
+      const [, data, jobOpts] = queue.add.mock.calls[0];
+      expect(data).toMatchObject({ fast: false });
+      expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
+    });
+  });
+
+  describe("addJob (child jobs)", () => {
+    it("sets lifo when data.fast is true", async () => {
+      const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
+      await (service as never as { addJob: (n: string, d: unknown) => Promise<void> }).addJob(
+        "prefetch-children",
+        { vehicleId: "v1", categoryId: "c1", source: "emex", action: "children", depth: 1, fast: true },
+      );
+      const [, , jobOpts] = queue.add.mock.calls[0];
+      expect(jobOpts).toMatchObject({ lifo: true });
+    });
+
+    it("omits lifo when data.fast is falsy", async () => {
+      const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
+      await (service as never as { addJob: (n: string, d: unknown) => Promise<void> }).addJob(
+        "prefetch-children",
+        { vehicleId: "v1", categoryId: "c1", source: "emex", action: "children", depth: 1 },
+      );
+      const [, , jobOpts] = queue.add.mock.calls[0];
+      expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
+    });
+  });
+
+  describe("processBackfillScan", () => {
+    it("runs Phase-1 in the fast lane even when the backlog is far over the ceiling", async () => {
+      // waiting 50k ≫ default maxBacklog 1000 → Phase-2 must be suspended, but a
+      // zero-parts vehicle must still be onboarded with lifo.
+      const { service, queue, redis } = makeDeps({
+        waiting: 50_000,
+        limitResults: [[{ id: "empty1", source: "emex" }]], // Phase-1 query result
+      });
+      await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
+
+      expect(queue.add).toHaveBeenCalledTimes(1);
+      const [name, data, jobOpts] = queue.add.mock.calls[0];
+      expect(name).toBe("prefetch-init");
+      expect(data).toMatchObject({ vehicleId: "empty1", fast: true });
+      expect(jobOpts).toMatchObject({ lifo: true });
+      // Phase-2 suspended → the rolling cursor is never read or advanced.
+      expect(redis.getJson).not.toHaveBeenCalled();
+      expect(redis.setJson).not.toHaveBeenCalled();
+    });
+
+    it("runs Phase-2 (normal lane, no lifo) when the backlog is under the ceiling", async () => {
+      // waiting 10 < 1000 → Phase-1 empty, Phase-2 picks a vehicle with fast=false.
+      const { service, queue, redis } = makeDeps({
+        waiting: 10,
+        limitResults: [
+          [], // Phase-1: no zero-parts vehicles
+          [{ id: "stale1", source: "emex", createdAt: new Date("2026-01-01T00:00:00Z") }], // Phase-2
+        ],
+      });
+      await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
+
+      expect(queue.add).toHaveBeenCalledTimes(1);
+      const [, data, jobOpts] = queue.add.mock.calls[0];
+      expect(data).toMatchObject({ vehicleId: "stale1", fast: false });
+      expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
+      // Phase-2 ran → cursor advanced.
+      expect(redis.setJson).toHaveBeenCalled();
+    });
+
+    it("skips entirely when the backfill gate is off (not prod host)", async () => {
+      process.env.CATALOG_BACKFILL_ENABLED = "false";
+      const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
+      await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
+      expect(queue.add).not.toHaveBeenCalled();
+    });
+  });
+});

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

apps/api/src/jobs/prefetch.types.ts

@@ -4,6 +4,13 @@ export type PrefetchSource = "pl24" | "emex" | "parts-catalogs";
 export interface PrefetchInitJobData {
   vehicleId: string;
   source: PrefetchSource;
+  /**
+   * Fast lane: enqueue this job (and its whole sub-job chain) with BullMQ `lifo`
+   * so it lands at the tail of the wait list and is picked before the deep
+   * deep-drill backlog. Set for Phase-1 (zero-parts) backfill so newly-decoded
+   * vehicles aren't starved behind the rolling rescan. See PrefetchWorkerService.
+   */
+  fast?: boolean;
 }
 
 /** Per-category job: fetches children OR parts */
@@ -13,4 +20,6 @@ export interface PrefetchCategoryJobData {
   source: PrefetchSource;
   action: "children" | "parts";
   depth: number;
+  /** Inherited from the init job — keeps the whole chain in the fast lane. */
+  fast?: boolean;
 }