diff --git a/apps/api/src/integrations/emex/emex.service.ts b/apps/api/src/integrations/emex/emex.service.ts index 11d6fd7..28294a2 100644 --- a/apps/api/src/integrations/emex/emex.service.ts +++ b/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 { + // 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 } diff --git a/apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts b/apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts index ba99657..e80cd8f 100644 --- a/apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts +++ b/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 { + // 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 } diff --git a/apps/api/src/integrations/pl24/pl24.service.ts b/apps/api/src/integrations/pl24/pl24.service.ts index 3819c13..7c151c4 100644 --- a/apps/api/src/integrations/pl24/pl24.service.ts +++ b/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 { + // 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 } diff --git a/apps/api/src/jobs/prefetch-context.ts b/apps/api/src/jobs/prefetch-context.ts new file mode 100644 index 0000000..38e3faa --- /dev/null +++ b/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:` 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 when executing inside a backfill / prefetch worker job. */ +export function isBackfillContext(): boolean { + return backfillContext.getStore() === true; +} diff --git a/apps/api/src/jobs/prefetch-utils.ts b/apps/api/src/jobs/prefetch-utils.ts index 3140cdf..8d1792c 100644 --- a/apps/api/src/jobs/prefetch-utils.ts +++ b/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 { 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"); } } diff --git a/apps/api/src/jobs/prefetch-worker.service.ts b/apps/api/src/jobs/prefetch-worker.service.ts index 0b7957e..eee236d 100644 --- a/apps/api/src/jobs/prefetch-worker.service.ts +++ b/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,