Merge pull request 'perf(backfill): stop the prefetch worker from throttling itself (Tier 1)' (#88) from dev into main

Reviewed-on: #88

apps/api/src/integrations/emex/emex.service.ts

@@ -20,6 +20,7 @@ import {
 import { ConfigService } from "@nestjs/config";
 
 import { ProxyAgent } from "undici";
+import { isBackfillContext } from "../../jobs/prefetch-context";
 import { RedisService } from "../../redis/redis.service";
 import { EmexBrowserService } from "./emex.browser";
 import { createEmptyDecodedVehicle, mapEmexResponse } from "./emex.mapper";
@@ -143,8 +144,11 @@ export class EmexService {
 
   /** Mark EMEX as actively used (5min TTL) to defer prefetch worker */
   private async touchActivity(): Promise<void> {
+    // Worker-originated fetches must NOT register as user activity, or the
+    // backfill worker throttles itself via checkCooldown.
+    if (isBackfillContext()) return;
     try {
-      await this.redis.set("prefetch:activity:emex", String(Date.now()), 300);
+      await this.redis.set("prefetch:activity:emex", String(Date.now()), 90);
     } catch {
       // Non-critical — don't break the request
     }

apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts

@@ -7,6 +7,7 @@
  */
 
 import { Injectable, Logger } from "@nestjs/common";
+import { isBackfillContext } from "../../jobs/prefetch-context";
 import { RedisService } from "../../redis/redis.service";
 import { PartsCatalogsAuthService } from "./parts-catalogs-auth.service";
 import {
@@ -43,8 +44,11 @@ export class PartsCatalogsService {
 
   /** Mark parts-catalogs as actively used (5min TTL) to defer prefetch worker */
   private async touchActivity(): Promise<void> {
+    // Worker-originated fetches must NOT register as user activity, or the
+    // backfill worker throttles itself via checkCooldown.
+    if (isBackfillContext()) return;
     try {
-      await this.redis.set("prefetch:activity:parts-catalogs", String(Date.now()), 300);
+      await this.redis.set("prefetch:activity:parts-catalogs", String(Date.now()), 90);
     } catch {
       // Non-critical
     }

apps/api/src/integrations/pl24/pl24.service.ts

@@ -14,6 +14,7 @@ import {
   ServiceUnavailableException,
 } from "@nestjs/common";
 import { ConfigService } from "@nestjs/config";
+import { isBackfillContext } from "../../jobs/prefetch-context";
 import { RedisService } from "../../redis/redis.service";
 import { StorageService } from "../../storage/storage.service";
 import { PL24AuthService } from "./pl24-auth.service";
@@ -827,8 +828,11 @@ export class PL24Service {
 
   /** Mark PL24 as actively used (5min TTL) to defer prefetch worker */
   private async touchActivity(): Promise<void> {
+    // Worker-originated fetches must NOT register as user activity, or the
+    // backfill worker throttles itself via checkCooldown.
+    if (isBackfillContext()) return;
     try {
-      await this.redis.set("prefetch:activity:pl24", String(Date.now()), 300);
+      await this.redis.set("prefetch:activity:pl24", String(Date.now()), 90);
     } catch {
       // Non-critical — don't break the request
     }

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

@@ -0,0 +1,21 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+
+/**
+ * Marks the current async call-chain as originating from the catalog backfill /
+ * prefetch worker.
+ *
+ * Source services (pl24 / emex / parts-catalogs) call `touchActivity()` on every
+ * fetch to set a `prefetch:activity:<source>` cooldown key, which the worker's
+ * `checkCooldown()` then honours by pausing. That key is meant to signal *user*
+ * activity so backfill yields to live users — but the worker's OWN fetches set it
+ * too, so the worker throttles itself (fetch → set 300s key → next job sees it →
+ * pause → repeat). Wrapping each worker job in this context lets `touchActivity()`
+ * skip the key when the call originates from the worker, so the cooldown reflects
+ * only real user requests.
+ */
+export const backfillContext = new AsyncLocalStorage<true>();
+
+/** True when executing inside a backfill / prefetch worker job. */
+export function isBackfillContext(): boolean {
+  return backfillContext.getStore() === true;
+}

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

@@ -41,15 +41,19 @@ export class RateLimitError extends Error {
 }
 
 /**
- * Check if a user is actively using the source.
- * Throws RateLimitError (1min retry) if cooldown key exists.
+ * Check if a *user* is actively using the source (the cooldown key is set only
+ * by user-facing requests; worker fetches skip it via the backfill context).
+ * Throws RateLimitError (retry after the key's remaining lifetime) if set.
  */
 export async function checkCooldown(redis: RedisService, source: string): Promise<void> {
   const key = `prefetch:activity:${source}`;
-  const exists = await redis.exists(key);
-  if (exists) {
-    const retryMs = source === "parts-catalogs" ? 120_000 : 60_000;
-    throw new RateLimitError(retryMs, "cooldown");
+  // ttl() returns the key's remaining lifetime in seconds (-2 = no key,
+  // -1 = no expiry). Pause for exactly that long so the worker waits once —
+  // until the real user-activity window ends — instead of re-checking every
+  // fixed 60s and burning ~5 empty cycles per key lifetime.
+  const ttl = await redis.ttl(key);
+  if (ttl > 0) {
+    throw new RateLimitError(ttl * 1000, "cooldown");
   }
 }
 

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

@@ -12,6 +12,7 @@ import { DATABASE, type Database } from "../database/database.provider";
 import { categories, parts, vehicles } from "../database/schema/core";
 import { RedisService } from "../redis/redis.service";
 import { QUEUE_NAMES, getBullConnection } from "./bull.config";
+import { backfillContext } from "./prefetch-context";
 import {
   RateLimitError,
   checkCooldown,
@@ -59,7 +60,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
   onModuleInit() {
     this.worker = new Worker(
       QUEUE_NAMES.CATALOG_PREFETCH,
-      (job, token) => this.process(job, token),
+      // Run every job inside the backfill context so source services' touchActivity()
+      // skips the user-activity cooldown key — the worker must not throttle itself.
+      (job, token) => backfillContext.run(true, () => this.process(job, token)),
       {
         connection: getBullConnection(),
         concurrency: 1,