Merge pull request 'Tam katalog kapsama: 4-faz prefetch stratejisi (derinlik+skip-completed+per-source+env)' (#146) from dev into main

2026-06-22 11:30:48 +03:00
parent a36008017c d119b1eb2c
commit fd884271b7
4 changed files with 190 additions and 20 deletions
--- a/apps/api/src/jobs/prefetch-utils.ts
+++ b/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";
--- a/apps/api/src/jobs/prefetch-worker.service.spec.ts
+++ b/apps/api/src/jobs/prefetch-worker.service.spec.ts
@@ -24,9 +24,13 @@ function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
    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
+    // 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
    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),
@@ -147,6 +151,24 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
      expect(redis.setJson).toHaveBeenCalled();
    });

+    it("skips a vehicle already marked complete (Phase-2 does not re-walk it)", async () => {
+      const { service, queue, redis } = makeDeps({
+        waiting: 10, // under ceiling → Phase-2 active
+        limitResults: [
+          [], // Phase-1: no zero-parts vehicles
+          [{ id: "done1", source: "emex", createdAt: new Date("2026-01-01T00:00:00Z") }], // Phase-2
+        ],
+      });
+      // scheduled guard absent (false), but the complete marker is present.
+      redis.exists.mockImplementation(async (...a: unknown[]) =>
+        String(a[0]).includes("prefetch:complete:"),
+      );
+      await (
+        service as never as { processBackfillScan: () => Promise<void> }
+      ).processBackfillScan();
+      expect(queue.add).not.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: [] });
@@ -175,4 +197,59 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
      expect(queue.add).not.toHaveBeenCalled();
    });
  });
+
+  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.
+    const nonLeaf = {
+      id: "c1",
+      linkPath: "/catalog/groups/x",
+      source: "pl24",
+      unavailable: false,
+      hasSubgroups: false,
+    };
+    type QCJ = {
+      queueCategoryJob: (c: unknown, v: string, s: string, d: number, f: boolean) => Promise<void>;
+    };
+
+    it("does NOT enqueue a children job at depth >= MAX_DEPTH (avoids a no-op)", async () => {
+      const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
+      // depth 12 == default MAX_DEPTH → processChildren would early-return, so the
+      // job must never be created.
+      await (service as never as QCJ).queueCategoryJob(nonLeaf, "v1", "pl24", 12, false);
+      expect(queue.add).not.toHaveBeenCalled();
+    });
+
+    it("enqueues a children job for a non-leaf within the cap", async () => {
+      // childCheck query (.limit(1)) returns no rows → enqueue a children fetch.
+      const { service, queue } = makeDeps({ waiting: 0, limitResults: [[]] });
+      await (service as never as QCJ).queueCategoryJob(nonLeaf, "v1", "pl24", 1, false);
+      expect(queue.add).toHaveBeenCalledTimes(1);
+      expect(queue.add.mock.calls[0][0]).toBe("prefetch-children");
+    });
+  });
 });
--- a/apps/api/src/jobs/prefetch-worker.service.ts
+++ b/apps/api/src/jobs/prefetch-worker.service.ts
@@ -27,17 +27,18 @@ import { PrefetchCategoryJobData, PrefetchInitJobData } from "./prefetch.types";
 import { CATALOG_PREFETCH_QUEUE } from "./queues/catalog-prefetch.queue";

 /**
- * How deep the background prefetch BFS drills the category tree.
+ * Depth cap for the prefetch BFS — a safety ceiling, NOT the target depth.
 *
- * Kept SHALLOW (2) on purpose: pre-warming the top ~2 levels for every decoded
- * vehicle is bounded work that finishes in days, kills empty catalog pages, and
- * makes the first clicks instant. Deeper categories are fetched lazily on-view
- * (getCategoryTree in CategoriesService) and cached permanently — so the long
- * tail nobody opens never costs an upstream request. Exhaustive depth-5
- * pre-warm produced an unbounded BFS fan-out (~460k jobs) that never drained
- * and starved the useful shallow + reactive work. Env-tunable.
+ * Goal is COMPLETE coverage: every decoded vehicle's full category tree + all
+ * parts in the DB so viewing is instant (DB read). The BFS terminates naturally
+ * at real leaves (isLeafLinkPath); this cap only guards against pathological /
+ * cyclic trees running away. Real PL24/EMEX/pcat trees are ~5-7 deep, so 12 is
+ * comfortably beyond every genuine leaf while still bounding a runaway.
+ * Efficiency is handled separately: queueCategoryJob never enqueues a children
+ * job past the cap (no early-return no-ops), and completed vehicles are skipped
+ * by the scan. Env-tunable (PREFETCH_MAX_DEPTH).
 */
-const MAX_DEPTH = Number(process.env.PREFETCH_MAX_DEPTH) || 2;
+const MAX_DEPTH = Number(process.env.PREFETCH_MAX_DEPTH) || 12;

 // ── Backfill scan (hourly cron) tuning ──
 /**
@@ -71,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.
@@ -89,6 +107,12 @@ const NORESULT_MAX_ATTEMPTS = Number(process.env.PREFETCH_NORESULT_MAX) || 2;
 /** TTL for the no-result counter — excluded vehicles retry after this, so a later
 *  catalog fix eventually re-fills them (default 7 days). */
 const NORESULT_TTL_S = (Number(process.env.PREFETCH_NORESULT_TTL_DAYS) || 7) * 86_400;
+/** TTL for the "fully fetched" marker. A vehicle whose whole prefetch chain
+ *  finished WITH parts is marked complete and skipped by the Phase-2 rescan, so
+ *  the scan stops re-walking finished trees (the old behaviour churned the queue
+ *  forever). The TTL re-validates periodically so a later upstream catalog change
+ *  is eventually picked up (default 21 days). Env-tunable. */
+const COMPLETE_TTL_S = (Number(process.env.PREFETCH_COMPLETE_TTL_DAYS) || 21) * 86_400;

@Injectable()
 export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
@@ -129,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
@@ -176,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));
      }
@@ -207,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;
@@ -473,6 +509,11 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
      // parts (no catalog data). They'd otherwise be re-picked every wave forever.
      const noResult = await this.redis.get(this.noResultKey(v.id));
      if (noResult && Number(noResult) >= NORESULT_MAX_ATTEMPTS) return;
+      // Skip vehicles whose full tree is already fetched (marker set on chain
+      // completion with parts). Stops Phase-2 from re-walking finished vehicles
+      // every wave; the marker's TTL re-validates them periodically. Phase-1
+      // (zero-parts) vehicles never carry this marker, so this is a no-op there.
+      if (await this.redis.exists(this.completeKey(v.id))) return;
      seen.add(v.id);
      picked.push({ id: v.id, source: v.source, fast });
    };
@@ -592,8 +633,11 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
          fast,
        });
      }
-    } else if (cat.linkPath) {
-      // Check if children already exist
+    } else 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
+      // slot on a no-op (this was ~85% of the queue at MAX_DEPTH=2).
      const [childCheck] = await this.db
        .select({ id: categories.id })
        .from(categories)
@@ -698,7 +742,15 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
        .from(parts)
        .where(eq(parts.vehicleId, vehicleId))
        .limit(1);
-      if (!hasPart) await this.markNoResult(vehicleId);
+      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);
+      } 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}`);
@@ -717,6 +769,29 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
    return `prefetch:noresult:${vehicleId}`;
  }

+  private completeKey(vehicleId: string): string {
+    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);
--- a/docker-compose.coolify.yml
+++ b/docker-compose.coolify.yml
@@ -116,6 +116,24 @@ services:
      - DIRECTUS_PROJECT=${DIRECTUS_PROJECT:-sase}
      # Bearer token for POST /blog/posts/internal (Süper Panel n8n pipeline)
      - BLOG_AUTOMATION_TOKEN=${BLOG_AUTOMATION_TOKEN:-}
+      # Catalog prefetch / backfill tuning (in-process worker lives in THIS api
+      # container, not worker.js). All empty → compiled-in defaults; set in Coolify
+      # to ramp without a code change. PL24 rate stays conservative unless rotating
+      # residential proxies are confirmed (upstream ban risk). See
+      # sase-coolify-compose-env-injection + sase-catalog-backfill memories.
+      - CATALOG_BACKFILL_ENABLED=${CATALOG_BACKFILL_ENABLED:-}
+      - PREFETCH_CONCURRENCY=${PREFETCH_CONCURRENCY:-}
+      - PREFETCH_RATE_MAX=${PREFETCH_RATE_MAX:-}
+      - PREFETCH_MAX_DEPTH=${PREFETCH_MAX_DEPTH:-}
+      - PREFETCH_RATE_PL24=${PREFETCH_RATE_PL24:-}
+      - PREFETCH_RATE_EMEX=${PREFETCH_RATE_EMEX:-}
+      - PREFETCH_RATE_PCAT=${PREFETCH_RATE_PCAT:-}
+      - PREFETCH_PCAT_DELAY_MS=${PREFETCH_PCAT_DELAY_MS:-}
+      - PREFETCH_PL24_START=${PREFETCH_PL24_START:-}
+      - PREFETCH_PL24_END=${PREFETCH_PL24_END:-}
+      - PREFETCH_NORESULT_MAX=${PREFETCH_NORESULT_MAX:-}
+      - PREFETCH_NORESULT_TTL_DAYS=${PREFETCH_NORESULT_TTL_DAYS:-}
+      - PREFETCH_COMPLETE_TTL_DAYS=${PREFETCH_COMPLETE_TTL_DAYS:-}
    depends_on:
      sase-redis:
        condition: service_healthy