From b886a05b5b5ba56b4b01e732b47e7da5be21ed81 Mon Sep 17 00:00:00 2001 From: Semih Yesilyurt Date: Wed, 10 Jun 2026 21:19:14 +0300 Subject: [PATCH] perf(pcat): rotating Floxy residential transport + resilience (decode latency) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prod decode was dominated by the DataImpulse proxy: ~50% of ports dead at any moment, 57% of captured tokens discarded by validation, the dead-port slot never evicted (invalidateSession only fired on 401/403, never on transport timeout), and a cold pool blocked the user inline on a multi-second capture. pcat sub-timing was p50 8.3s / p90 56s, blowing the 25s decode budget. Verified on prod: the TWS- token is NOT request-time IP-bound (0 auth 401/403 across thousands of calls; a token captured on one IP replays 200 through arbitrary fresh residential IPs). So the proxy can rotate freely. - Transport swap to Floxy residential (PCAT_PROXY_PROVIDER=floxy|dataimpulse|none). Call/validate leg = a fresh exit IP per request (max IP diversity → min per-IP ban); capture leg = a sticky session per Playwright attempt. DataImpulse + direct kept as env fallbacks. Token decoupled from the slot (proxyPort removed). - Prune 3 zero-yield JWT sites (knkauto/autodo.kz/flynestauto — 0 captures, ~80% of attempts); add PCAT_JWT_SITES env override. - Lower catalog-scrape timeout 30s→10s (env-tunable) so a slow IP aborts within budget and the retry rotates to a fresh IP. - Cold-pool fast-fail on the decode race: warm in background, let EMEX/PL24 answer instead of blocking inline; treated as a transient miss (no neg-cache). - Evict a token after N consecutive transport failures on rotating IPs (the dead-slot bug); reset on any 2xx. - PCAT circuit breaker mirroring PL24 (transient-only; cold-skip neutral; skip marks outcome.transient so a pcat-only VIN isn't negative-cached 30m). Typecheck + Biome + unit tests (categories/vehicles) green. Default flips pcat to Floxy on next deploy; instant rollback via PCAT_PROXY_PROVIDER. Co-Authored-By: Claude Fable 5 --- .../parts-catalogs-auth.service.ts | 291 +++++++++++++----- .../parts-catalogs/parts-catalogs.service.ts | 109 +++++-- .../parts-catalogs/parts-catalogs.types.ts | 13 +- apps/api/src/vehicles/vehicles.service.ts | 77 ++++- 4 files changed, 373 insertions(+), 117 deletions(-) diff --git a/apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts b/apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts index adea463..2dd2115 100644 --- a/apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts +++ b/apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts @@ -1,12 +1,16 @@ /** - * Parts-Catalogs Auth Service — v3 token warm pool via Playwright + DataImpulse proxy + * Parts-Catalogs Auth Service — v3 token warm pool via Playwright + rotating proxy * * Tokens are captured by navigating to partner sites that embed the v3 widget. * The widget calls /v3/api/proxy/* with `x-api-key: TWS-{UUID}` and four other * X-* headers (api-path, gui-version, user-id, origin, referer); we intercept * all of them so backend requests can replay the exact header set. * - * Tokens are IP-bound — same proxy port must be reused for the API calls. + * Transport (PCAT_PROXY_PROVIDER): "floxy" residential (default) rotates a fresh + * exit IP per call (the token is NOT IP-bound — see parts-catalogs.types.ts — so + * max IP diversity = min per-IP ban), with a sticky IP only per Playwright + * capture; "dataimpulse" is the legacy port-bound datacenter proxy; "none" calls + * direct from the server IP. * * Warm pool behavior: * 24/7 → proactive: maintain >= 1 slot, auto-refresh ~9.5min before expiry. @@ -22,10 +26,10 @@ import { RedisService } from "../../redis/redis.service"; import { JwtSlot, PcatJwtToken, PcatSession } from "./parts-catalogs.types"; // Shared single-slot cache so dev + prod (and any restarted container) can -// inherit a freshly-captured warm slot instead of paying the Playwright -// capture cost from cold. Token is still IP-bound to its proxyPort; if a -// reader on a different IP gets 401/403, invalidateSession's existing -// fallback re-captures locally. So worst case = today's behavior. +// inherit a freshly-captured warm token instead of paying the Playwright +// capture cost from cold. The token is portable (not IP-bound), so any +// container can serve it through its own rotating call-leg IPs; if it's dead +// at the source, invalidateSession drops it and the next read re-captures. const REDIS_SLOT_KEY = "pcat:jwt:slot"; const REDIS_SAFETY_BUFFER_S = 60; // don't serve from Redis if <60s of life left @@ -62,30 +66,46 @@ const RPM_PER_SLOT = 6; // 1 token per 6 req/min * Each site uses a different proxy port (IP) to avoid rate limiting. */ // Order matters: round-robin starts from index 0, so the first site is the one -// a cold pool waits on. Sites that consistently load in <10s through the proxy -// stay at the top so initial JWT capture finishes in seconds. -// (e-acca.com was previously included last but removed — the current DataImpulse -// rotating proxy 74.81.81.81:10000-10999 cannot reach it; page.goto always -// blocked until PAGE_TIMEOUT instead of failing fast like the other sites.) -const JWT_SITES = [ +// a cold pool waits on. Sites that consistently load fast stay at the top so +// initial JWT capture finishes in seconds. +// Pruned 2026-06-10 to the 6 sites that actually yielded captures on prod — +// knkauto.ru / autodo.kz / flynestauto.com produced 0 captures yet consumed +// ~80% of capture attempts (the per-site cooldown kept forcing the retry loop +// onto them). Override the list at runtime with PCAT_JWT_SITES (comma-separated) +// without a deploy, e.g. to give dev a disjoint subset. +const DEFAULT_JWT_SITES = [ "https://www.alkatalog.com/cats/#/catalogs", "https://auto-komplekt.ru/goodvin#/catalogs", "https://www.autotrade.md/cats/#/catalogs", "https://www.e-trak.ru/cats/#/catalogs", "https://www.autopolyus.ru/cats/#/catalogs", - "https://knkauto.ru/goodvin#/catalogs", - "https://www.autodo.kz/#/catalogs", "https://avtoman124.ru/goodvin#/catalogs", - "https://flynestauto.com/auto-parts-oem-catalog", ]; -// DataImpulse proxy defaults (port-based IP rotation) +// DataImpulse proxy defaults (port-based IP rotation) — legacy. Half its ports +// were dead at any moment (measured 50% connect-timeout), which drove the bulk of +// pcat decode latency + failures. Kept selectable via PCAT_PROXY_PROVIDER for +// instant rollback, but Floxy is the default. const DI_HOST = "gw.dataimpulse.com"; const DI_PORT_MIN = 10000; const DI_PORT_MAX = 10999; const DI_DEFAULT_USER = "1726bbe361918676d44e"; const DI_DEFAULT_PASS = "78ebc3d881de6ec0"; +// Floxy residential defaults (session-based IP rotation; same account EMEX uses). +// Call leg → plain creds = a fresh residential exit IP per request (max IP +// diversity → minimal per-IP ban exposure, since each IP sees ~1 request). +// Capture leg → a sticky session id pins one IP for the duration of a single +// Playwright capture so the partner site sees a coherent session. +const FLOXY_HOST = "residential.floxy.io"; +const FLOXY_PORT = 12321; +const FLOXY_DEFAULT_USER = "d739255e819b"; +const FLOXY_DEFAULT_PASS = "9092873ba4e0"; +const FLOXY_CAPTURE_LIFETIME_S = 60; // sticky window for one capture attempt + +type ProxyProvider = "floxy" | "dataimpulse" | "none"; +type ProxyLeg = "call" | "capture"; + /** Simple counting semaphore (same pattern as EmexBrowserService) */ class Semaphore { private current = 0; @@ -119,18 +139,20 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { private browser: Browser | null = null; private launching: Promise | null = null; - // Concurrent captures capped at JWT_SITES.length — the real bottleneck is - // how many distinct partner sites we can drive in parallel (one capture per - // site, since each site = a different IP / cookie origin). Ports are - // ~unlimited (10000-10999) and Playwright contexts are isolated; the - // synchronous siteIndex round-robin hands each concurrent capture a - // different site so they don't collide. - private readonly semaphore = new Semaphore(JWT_SITES.length); + // Active JWT capture sites (PCAT_JWT_SITES override or DEFAULT_JWT_SITES), + // resolved in the constructor. + private readonly jwtSites: string[]; + // Concurrent captures capped at jwtSites.length — the real bottleneck is how + // many distinct partner sites we can drive in parallel (one capture per site, + // since each site = a different cookie origin). Playwright contexts are + // isolated; the synchronous siteIndex round-robin hands each concurrent + // capture a different site so they don't collide. Initialized in constructor. + private readonly semaphore: Semaphore; // Pool state private pool: JwtSlot[] = []; private siteLastUsedAt = new Map(); - private siteIndex = 0; // round-robin across JWT_SITES + private siteIndex = 0; // round-robin across jwtSites private requestRoundRobin = 0; // round-robin across pool slots // Business hours scheduling @@ -139,20 +161,108 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { // RPM tracking private requestTimestamps: number[] = []; - // Config - private readonly useProxy: boolean; - private readonly proxyHost: string; - private readonly proxyUser: string; - private readonly proxyPass: string; + // Transport config — see ProxyProvider. The TWS- token is NOT request-time + // IP-bound (parts-catalogs.types.ts), so the call leg rotates exit IPs per + // request and only the capture leg holds a sticky IP per attempt. + private readonly proxyProvider: ProxyProvider; + // Floxy residential + private readonly floxyHost: string; + private readonly floxyPort: number; + private readonly floxyUser: string; + private readonly floxyPass: string; + private readonly floxyCaptureLifetime: number; + // DataImpulse legacy (port-rotated) + private readonly diHost: string; + private readonly diUser: string; + private readonly diPass: string; constructor( private configService: ConfigService, private readonly redis: RedisService, ) { - this.useProxy = this.configService.get("PCAT_USE_PROXY", "true") === "true"; - this.proxyHost = this.configService.get("PCAT_PROXY_HOST", DI_HOST); - this.proxyUser = this.configService.get("PCAT_PROXY_USER", DI_DEFAULT_USER); - this.proxyPass = this.configService.get("PCAT_PROXY_PASS", DI_DEFAULT_PASS); + const cfg = this.configService; + + // Active capture sites: PCAT_JWT_SITES override (comma-separated) or the + // pruned default. Empty/blank entries are dropped; falls back to default if + // the override resolves to nothing. + const sitesRaw = cfg.get("PCAT_JWT_SITES", ""); + const parsedSites = sitesRaw + .split(",") + .map((s) => s.trim()) + .filter((s) => s.length > 0); + this.jwtSites = parsedSites.length > 0 ? parsedSites : DEFAULT_JWT_SITES; + this.semaphore = new Semaphore(this.jwtSites.length); + + // Provider selection. Back-compat: PCAT_USE_PROXY=false still forces direct. + const rawProvider = cfg.get("PCAT_PROXY_PROVIDER", "floxy").toLowerCase(); + const useProxy = cfg.get("PCAT_USE_PROXY", "true") === "true"; + this.proxyProvider = !useProxy + ? "none" + : rawProvider === "dataimpulse" || rawProvider === "none" + ? (rawProvider as ProxyProvider) + : "floxy"; + + const toInt = (key: string, def: number): number => { + const n = Number(cfg.get(key, def)); + return Number.isInteger(n) && n >= 0 ? n : def; + }; + const toPort = (key: string, def: number): number => { + const n = Number(cfg.get(key, def)); + return Number.isInteger(n) && n >= 1 && n <= 65535 ? n : def; + }; + + this.floxyHost = cfg.get("PCAT_FLOXY_HOST", FLOXY_HOST); + this.floxyPort = toPort("PCAT_FLOXY_PORT", FLOXY_PORT); + this.floxyUser = cfg.get("PCAT_FLOXY_USER", FLOXY_DEFAULT_USER); + this.floxyPass = cfg.get("PCAT_FLOXY_PASS", FLOXY_DEFAULT_PASS); + this.floxyCaptureLifetime = toInt("PCAT_FLOXY_CAPTURE_LIFETIME", FLOXY_CAPTURE_LIFETIME_S); + + this.diHost = cfg.get("PCAT_PROXY_HOST", DI_HOST); + this.diUser = cfg.get("PCAT_PROXY_USER", DI_DEFAULT_USER); + this.diPass = cfg.get("PCAT_PROXY_PASS", DI_DEFAULT_PASS); + + this.logger.log( + this.proxyProvider === "floxy" + ? `PCAT transport: floxy residential ${this.floxyHost}:${this.floxyPort} (call=rotating, capture=sticky ${this.floxyCaptureLifetime}s)` + : this.proxyProvider === "dataimpulse" + ? `PCAT transport: dataimpulse ${this.diHost}:${DI_PORT_MIN}-${DI_PORT_MAX}` + : "PCAT transport: direct (no proxy)", + ); + } + + /** + * Build a proxy for one leg. "call" rotates the exit IP per invocation (the + * token is not IP-bound, so each decode/scrape/validate request gets a fresh + * residential IP → max IP diversity, min ban). "capture" pins a sticky IP for + * the lifetime of a single Playwright capture so the partner site sees a + * coherent session. Returns nulls when provider is "none" (direct). + */ + private buildProxy(leg: ProxyLeg): { + proxyUrl: string | null; + proxyConfig: { server: string; username: string; password: string } | null; + } { + if (this.proxyProvider === "none") return { proxyUrl: null, proxyConfig: null }; + + if (this.proxyProvider === "dataimpulse") { + const port = this.allocatePort(); + const server = `http://${this.diHost}:${port}`; + return { + proxyUrl: `http://${this.diUser}:${this.diPass}@${this.diHost}:${port}`, + proxyConfig: { server, username: this.diUser, password: this.diPass }, + }; + } + + // floxy + let password = this.floxyPass; + if (leg === "capture" && this.floxyCaptureLifetime > 0) { + const sid = Math.random().toString(36).slice(2, 10); + password = `${this.floxyPass}_session-${sid}_lifetime-${this.floxyCaptureLifetime}`; + } + const server = `http://${this.floxyHost}:${this.floxyPort}`; + return { + proxyUrl: `http://${this.floxyUser}:${password}@${this.floxyHost}:${this.floxyPort}`, + proxyConfig: { server, username: this.floxyUser, password }, + }; } async onModuleInit(): Promise { @@ -193,10 +303,14 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { // ─── Public API ─────────────────────────────────────────── /** - * Acquire a matched JWT + proxy session from the pool. - * Returns a valid slot (round-robin) or captures on-demand if pool is empty. + * Acquire a JWT session from the pool. + * Returns a valid slot (round-robin) or, if the pool is empty, captures + * on-demand. With `blockOnEmpty: false` it does NOT block on a cold pool — + * it kicks off a background warm-up and returns null so a latency-sensitive + * caller (the decode race) can let a faster source answer instead of eating + * the multi-second capture cost inline. */ - async acquireSession(): Promise { + async acquireSession(opts?: { blockOnEmpty?: boolean }): Promise { this.trackRequest(); // Try to find a valid slot in the pool @@ -207,14 +321,26 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { return this.slotToSession(slot); } - // No valid slot — capture on-demand + // No valid slot. Latency-sensitive callers fast-fail and warm in background. + if (opts?.blockOnEmpty === false) { + this.logger.log("JWT pool empty — fast-fail (warming in background)"); + this.captureToPool().catch((err) => { + this.logger.warn(`Background warm capture failed: ${(err as Error).message}`); + }); + return null; + } + + // Capture on-demand (blocks the caller). this.logger.log("JWT pool empty — capturing on-demand..."); const newSlot = await this.captureToPool(); return this.slotToSession(newSlot); } /** - * Invalidate a session after 401/403, remove the slot, and capture a replacement. + * Invalidate a session (401/403, or N consecutive transport failures on + * rotating IPs = a token that's dead at the source), remove the slot, and + * capture a replacement only if the pool is now empty — a surviving sibling + * plus the scheduled refresh/scale-up covers the rest without a capture storm. */ async invalidateSession(session: PcatSession): Promise { const idx = this.pool.indexOf(session._slot); @@ -222,17 +348,18 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { this.clearSlotTimer(this.pool[idx]); this.pool.splice(idx, 1); this.logger.log( - `JWT pool: slot invalidated (port ${session._slot.proxyPort}), ${this.pool.length} remaining`, + `JWT pool: slot invalidated (${new URL(session._slot.siteUsed).hostname}), ${this.pool.length} remaining`, ); } - // The Redis-shared slot may point at a now-401'd JWT (e.g. IP rotated - // away on the proxy). Drop it so other containers re-capture instead of - // inheriting the same dead slot. + // The Redis-shared slot may point at this same now-dead token. Drop it so + // a cold-booting container re-captures instead of hydrating the dead token. this.deleteFromRedis().catch(() => {}); - // Capture replacement in background (don't block the caller's retry) - if (this.isBusinessHours() || this.pool.length === 0) { + // Only force an immediate replacement when the pool emptied; otherwise let + // the surviving slot serve and the refresh/scale-up timers backfill, so a + // burst of evictions can't stampede the capture pipeline. + if (this.pool.length === 0) { this.captureToPool().catch((err) => { this.logger.error(`Replacement capture failed: ${(err as Error).message}`); }); @@ -262,16 +389,10 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { } private slotToSession(slot: JwtSlot): PcatSession { - const proxyUrl = this.useProxy - ? `http://${this.proxyUser}:${this.proxyPass}@${this.proxyHost}:${slot.proxyPort}` - : null; - const proxyConfig = this.useProxy - ? { - server: `http://${this.proxyHost}:${slot.proxyPort}`, - username: this.proxyUser, - password: this.proxyPass, - } - : null; + // Fresh rotating call-leg proxy per request — and per retry, since + // fetchWithAuth re-acquires the session each attempt → a new exit IP each + // time, so a slow/blocked residential IP is shed on the next try. + const { proxyUrl, proxyConfig } = this.buildProxy("call"); return { apiKey: slot.jwt.raw, apiPath: slot.jwt.apiPath, @@ -294,17 +415,16 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { const maxRetries = 4; for (let attempt = 0; attempt < maxRetries; attempt++) { - const port = this.allocatePort(); const siteUrl = this.getAvailableSite(); - const jwt = await this.attemptCapture(siteUrl, port); + const jwt = await this.attemptCapture(siteUrl); if (jwt) { - // Validate the JWT+proxy combo against the real upstream before - // publishing the slot — see VALIDATION_VIN comment above. - const valid = await this.validateSlot(jwt, port); + // Validate the token against the real upstream (on a fresh rotating + // call-leg IP) before publishing the slot — see VALIDATION_VIN. + const valid = await this.validateSlot(jwt); if (!valid) { this.logger.warn( - `JWT capture attempt ${attempt + 1}/${maxRetries} validation failed (port ${port}, ${new URL(siteUrl).hostname}), trying next site...`, + `JWT capture attempt ${attempt + 1}/${maxRetries} validation failed (${new URL(siteUrl).hostname}), trying next site...`, ); // Mark site as used anyway so we rotate away rather than retry it. this.siteLastUsedAt.set(siteUrl, Date.now()); @@ -313,10 +433,10 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { const slot: JwtSlot = { jwt, - proxyPort: port, siteUsed: siteUrl, capturedAt: Date.now(), refreshTimer: null, + failCount: 0, }; // Mark site as used @@ -352,13 +472,14 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { } /** - * Probe the just-captured token + proxy port with a single cheap upstream - * call (/car/info with a public demo VIN). Returns true iff the upstream - * accepts the combo with 2xx within VALIDATION_TIMEOUT_MS. A failure here - * (proxy dead, IP blocked, token 401/403) is far cheaper to absorb at - * capture time than to inherit when a real user clicks a category. + * Probe the just-captured token with a single cheap upstream call (/car/info + * with a public demo VIN) on a fresh rotating call-leg IP — the SAME transport + * a real request will use. Returns true iff upstream accepts it with 2xx within + * VALIDATION_TIMEOUT_MS. Validating on the call-leg transport (not the capture + * IP) is what makes the check meaningful: it rejects a token only if it can't + * be served the way it will actually be served. */ - private async validateSlot(jwt: PcatJwtToken, port: number): Promise { + private async validateSlot(jwt: PcatJwtToken): Promise { const url = `${PCAT_API_BASE}/car/info?q=${VALIDATION_VIN}`; const fetchOptions: RequestInit & { dispatcher?: any } = { method: "GET", @@ -375,10 +496,11 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { }, signal: AbortSignal.timeout(VALIDATION_TIMEOUT_MS), }; - if (this.useProxy) { + const { proxyUrl } = this.buildProxy("call"); + if (proxyUrl) { const { ProxyAgent } = await import("undici"); fetchOptions.dispatcher = new ProxyAgent({ - uri: `http://${this.proxyUser}:${this.proxyPass}@${this.proxyHost}:${port}`, + uri: proxyUrl, connect: { timeout: 6_000 }, }); } @@ -539,15 +661,16 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { private getAvailableSite(): string { const now = Date.now(); + const sites = this.jwtSites; // Try round-robin, preferring sites not on cooldown - for (let i = 0; i < JWT_SITES.length; i++) { - const idx = (this.siteIndex + i) % JWT_SITES.length; - const site = JWT_SITES[idx]; + for (let i = 0; i < sites.length; i++) { + const idx = (this.siteIndex + i) % sites.length; + const site = sites[idx]; const lastUsed = this.siteLastUsedAt.get(site) || 0; if (now - lastUsed >= SITE_COOLDOWN) { - this.siteIndex = (idx + 1) % JWT_SITES.length; + this.siteIndex = (idx + 1) % sites.length; return site; } } @@ -555,15 +678,15 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { // All on cooldown — pick the one with oldest usage let oldestIdx = 0; let oldestTime = Number.POSITIVE_INFINITY; - for (let i = 0; i < JWT_SITES.length; i++) { - const lastUsed = this.siteLastUsedAt.get(JWT_SITES[i]) || 0; + for (let i = 0; i < sites.length; i++) { + const lastUsed = this.siteLastUsedAt.get(sites[i]) || 0; if (lastUsed < oldestTime) { oldestTime = lastUsed; oldestIdx = i; } } - this.siteIndex = (oldestIdx + 1) % JWT_SITES.length; - return JWT_SITES[oldestIdx]; + this.siteIndex = (oldestIdx + 1) % sites.length; + return sites[oldestIdx]; } private allocatePort(): number { @@ -572,19 +695,17 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy { // ─── JWT capture via Playwright ─────────────────────────── - private async attemptCapture(siteUrl: string, port: number): Promise { + private async attemptCapture(siteUrl: string): Promise { let context: BrowserContext | null = null; const startTime = Date.now(); try { - // Build context options with proxy + // Sticky capture proxy: one residential exit IP pinned for the lifetime of + // this capture so the partner site + widget see a coherent session. const contextOptions: Record = {}; - if (this.useProxy) { - contextOptions.proxy = { - server: `http://${this.proxyHost}:${port}`, - username: this.proxyUser, - password: this.proxyPass, - }; + const { proxyConfig } = this.buildProxy("capture"); + if (proxyConfig) { + contextOptions.proxy = proxyConfig; } if (!this.browser) throw new Error("PCAT browser not initialized"); 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 713e953..dbe22c9 100644 --- a/apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts +++ b/apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts @@ -2,8 +2,9 @@ * Parts-Catalogs API Service — HTTP client for parts-catalogs.com v3 * * Calls the v3 widget proxy (gui.parts-catalogs.com/v3/api/proxy/*) with the - * captured TWS- token + supporting X-* headers. Requests must go through the - * same DataImpulse proxy port that captured the token (IP-bound). + * captured TWS- token + supporting X-* headers. The token is portable (NOT + * IP-bound), so each request rotates a fresh exit IP via the auth service's + * call-leg proxy (PartsCatalogsAuthService.buildProxy). */ import { Injectable, Logger } from "@nestjs/common"; @@ -20,19 +21,38 @@ import { } from "./parts-catalogs.types"; const API_BASE = "https://gui.parts-catalogs.com/v3/api/proxy"; -const REQUEST_TIMEOUT = Number(process.env.PCAT_REQUEST_TIMEOUT_MS) || 30_000; +// Catalog scrape (groups2/parts2). Lowered 30s→10s (2026-06-10): the old 30s +// default could blow the caller's 25s decode budget on a single slow attempt, +// and through a rotating residential proxy a healthy call answers in 1-4s. A +// slow/blocked exit IP now aborts at 10s and the retry re-acquires a fresh IP. +const REQUEST_TIMEOUT = Number(process.env.PCAT_REQUEST_TIMEOUT_MS) || 10_000; const MAX_RETRIES = Number(process.env.PCAT_MAX_RETRIES) || 2; -// A dead DataImpulse proxy port otherwise stalls for undici's 10s default connect -// timeout before the retry rotates to a fresh port — three of those blow the -// caller's 25s decode budget. Fail fast so retries reach a live port in time. +// undici's ProxyAgent connect.timeout bounds the connect to the proxy entry. const PROXY_CONNECT_TIMEOUT = Number(process.env.PCAT_PROXY_CONNECT_TIMEOUT_MS) || 6_000; -// VIN decode (/car/info) answers in <1s on a healthy proxy, so a long timeout only -// prolongs dead-port connects. undici's ProxyAgent connect.timeout does NOT bound the -// connection to the proxy itself (it stays at undici's 10s default), so we bound it at -// the fetch level: a short per-call timeout + extra retries makes a stuck DataImpulse -// port abort fast and rotate to a live one within the caller's 25s decode budget. +// VIN decode (/car/info) answers in 1-4s through the rotating residential proxy. +// A short per-call timeout + retries means a slow exit IP aborts fast and the +// next attempt re-acquires the session → a fresh IP, all within the 25s budget. const DECODE_REQUEST_TIMEOUT = Number(process.env.PCAT_DECODE_TIMEOUT_MS) || 6_000; const DECODE_MAX_RETRIES = Number(process.env.PCAT_DECODE_MAX_RETRIES) || 4; +// Evict a pooled token after this many consecutive transport failures (reset on +// any 2xx). Through the rotating residential proxy each attempt hits a different +// exit IP, so N failures in a row on N different IPs ⇒ the token is dead at the +// source, not a flaky IP. Pre-rotation this was the "dead slot never evicted" +// bug: invalidateSession only fired on 401/403, so a bad token was served until TTL. +const TRANSPORT_FAIL_EVICT_THRESHOLD = Number(process.env.PCAT_TRANSPORT_FAIL_EVICT) || 2; + +/** + * Thrown when a fast-fail (blockOnEmpty:false) caller hits a cold pool. NOT a + * "VIN not found" — pcat was skipped while the pool warmed, so the caller must + * treat it as a transient miss (don't poison the negative cache) and let + * another source answer. + */ +class PcatColdPoolError extends Error { + constructor() { + super("pcat cold pool — skipped to avoid inline capture"); + this.name = "PcatColdPoolError"; + } +} @Injectable() export class PartsCatalogsService { @@ -62,7 +82,8 @@ export class PartsCatalogsService { async decodeVin( vin: string, signal?: AbortSignal, - outcome?: { transient: boolean }, + outcome?: { transient: boolean; coldSkip?: boolean }, + opts?: { blockOnEmpty?: boolean }, ): Promise { if (!(await this.posthog.isSourceLive("parts-catalogs"))) { this.logger.warn("parts-catalogs disabled by kill switch (kill-source-parts-catalogs)"); @@ -72,6 +93,7 @@ export class PartsCatalogsService { const data = await this.fetchWithAuth("/car/info", { q: vin }, signal, { timeoutMs: DECODE_REQUEST_TIMEOUT, maxRetries: DECODE_MAX_RETRIES, + blockOnEmpty: opts?.blockOnEmpty, }); if (!data || typeof data !== "object") { @@ -105,9 +127,23 @@ export class PartsCatalogsService { return { cars }; } catch (err) { const e = err as Error & { cause?: unknown }; - // Distinguish a transient transport blip (proxy timeout / dropped DataImpulse - // connection) from a genuine "not found". The caller uses `outcome.transient` - // to decide whether to poison the negative cache — a blip must not. + // Cold-pool fast-fail: pcat was skipped (pool warming), NOT a definitive + // miss. Mark transient so the orchestrator doesn't negative-cache a VIN + // pcat hasn't actually evaluated, and stay quiet (this is expected). + if (e instanceof PcatColdPoolError) { + if (outcome) { + outcome.transient = true; + // A deliberate cold-pool skip is not a transport fault — flag it so the + // circuit breaker doesn't count it (the pool is already warming). + outcome.coldSkip = true; + } + this.logger.debug(`pcat decode skipped for ${vin}: cold pool (warming)`); + return null; + } + // Distinguish a transient transport blip (proxy timeout / dropped + // connection) from a genuine "not found". The caller uses + // `outcome.transient` to decide whether to poison the negative cache — a + // blip must not. if ( outcome && (e.name === "TimeoutError" || @@ -212,7 +248,7 @@ export class PartsCatalogsService { endpoint: string, params?: Record, externalSignal?: AbortSignal, - opts?: { timeoutMs?: number; maxRetries?: number }, + opts?: { timeoutMs?: number; maxRetries?: number; blockOnEmpty?: boolean }, ): Promise { const maxRetries = opts?.maxRetries ?? MAX_RETRIES; const timeoutMs = opts?.timeoutMs ?? REQUEST_TIMEOUT; @@ -223,7 +259,12 @@ export class PartsCatalogsService { if (externalSignal?.aborted) { throw new Error("Request aborted by caller"); } - session = await this.authService.acquireSession(); + session = await this.authService.acquireSession({ blockOnEmpty: opts?.blockOnEmpty }); + if (!session) { + // Cold pool + fast-fail: warming in the background; don't retry, let the + // caller fall through to another source. + throw new PcatColdPoolError(); + } const url = new URL(`${API_BASE}${endpoint}`); if (params) { @@ -263,6 +304,8 @@ export class PartsCatalogsService { const response = await fetch(url.toString(), fetchOptions); if (response.ok) { + // Token proven good on this IP — clear any accrued failure streak. + session._slot.failCount = 0; return await response.json(); } @@ -291,16 +334,28 @@ export class PartsCatalogsService { /fetch failed|ECONNRESET|ECONNREFUSED|EAI_AGAIN|socket hang up|other side closed|terminated|UND_ERR/i.test( `${e.message} ${String(e.cause ?? "")}`, ); - if (transient && attempt < maxRetries) { - this.logger.warn( - `Transient transport error on ${endpoint} (attempt ${attempt + 1}/${maxRetries + 1}): ${e.message}${ - e.cause ? ` [cause: ${String(e.cause)}]` : "" - }`, - ); - // Brief backoff so a momentarily-flaky proxy port can recover; the next - // loop iteration re-acquires a session (round-robin across the pool). - await new Promise((r) => setTimeout(r, 400 * (attempt + 1))); - continue; + if (transient) { + // A token that keeps failing transport across rotating IPs is dead at + // the source — evict it so it stops being round-robin-served (the old + // 401/403-only eviction never caught this). Only when a proxy is in use; + // direct-mode transport errors are upstream, not token/IP problems. + if (session.proxyUrl) { + session._slot.failCount = (session._slot.failCount ?? 0) + 1; + if (session._slot.failCount >= TRANSPORT_FAIL_EVICT_THRESHOLD) { + await this.authService.invalidateSession(session); + } + } + if (attempt < maxRetries) { + this.logger.warn( + `Transient transport error on ${endpoint} (attempt ${attempt + 1}/${maxRetries + 1}): ${e.message}${ + e.cause ? ` [cause: ${String(e.cause)}]` : "" + }`, + ); + // Brief backoff so a momentarily-flaky exit IP can recover; the next + // loop iteration re-acquires a session (a fresh rotating IP). + await new Promise((r) => setTimeout(r, 400 * (attempt + 1))); + continue; + } } throw err; } diff --git a/apps/api/src/integrations/parts-catalogs/parts-catalogs.types.ts b/apps/api/src/integrations/parts-catalogs/parts-catalogs.types.ts index 65a2fe1..e0798c9 100644 --- a/apps/api/src/integrations/parts-catalogs/parts-catalogs.types.ts +++ b/apps/api/src/integrations/parts-catalogs/parts-catalogs.types.ts @@ -1,7 +1,12 @@ /** * Captured from a parts-catalogs.com v3 widget request. * Token + the supporting X-* headers the widget sends with every API call. - * IP-bound (must be reused with the same proxy port that captured it). + * + * NOT request-time IP-bound. Verified on prod 2026-06-10: 0 auth 401/403 across + * thousands of proxied calls, and a token captured on one exit IP replays with a + * 200 through arbitrary other IPs (incl. fresh residential ones). Upstream issues + * a stable shared token per origin. So the call leg rotates exit IPs freely; only + * the capture (Playwright) leg holds a sticky IP for the duration of one capture. */ export interface PcatJwtToken { raw: string; // x-api-key value, e.g. "TWS-016EA7BE-..." @@ -15,10 +20,14 @@ export interface PcatJwtToken { export interface JwtSlot { jwt: PcatJwtToken; - proxyPort: number; siteUsed: string; capturedAt: number; refreshTimer: ReturnType | null; + // Consecutive transport failures across requests (reset on any 2xx). The slot + // is evicted once this crosses a threshold — a token that keeps failing on + // fresh rotating IPs is dead at the source, not a flaky exit IP. Optional so + // slots persisted to Redis by older builds still hydrate. + failCount?: number; } export interface PcatSession { diff --git a/apps/api/src/vehicles/vehicles.service.ts b/apps/api/src/vehicles/vehicles.service.ts index 5f0888b..044a73d 100644 --- a/apps/api/src/vehicles/vehicles.service.ts +++ b/apps/api/src/vehicles/vehicles.service.ts @@ -29,7 +29,7 @@ import { import { EmexService } from "../integrations/emex/emex.service"; import { EmexCandidate } from "../integrations/emex/emex.service"; import { PartsCatalogsService } from "../integrations/parts-catalogs/parts-catalogs.service"; -import { PcatCar } from "../integrations/parts-catalogs/parts-catalogs.types"; +import { PcatCar, PcatVinResult } from "../integrations/parts-catalogs/parts-catalogs.types"; import { PL24Service } from "../integrations/pl24/pl24.service"; import { VinApiService } from "../integrations/vin-api/vin-api.service"; import { PrefetchSource } from "../jobs/prefetch.types"; @@ -488,6 +488,47 @@ export class VehiclesService { await this.redis.del(VehiclesService.PL24_CB_FAILURE_KEY); } + // ─── PCAT circuit breaker ────────────────────────────── + // Mirrors the PL24 breaker. parts-catalogs is the #1 winning decode source, so + // when its transport degrades (proxy/upstream bad window) it drags the whole + // race. After N consecutive TRANSIENT pcat failures we open the circuit briefly + // and skip pcat so EMEX/PL24 answer fast. CRITICAL: only transient faults count + // — a definitive "no cars" miss must NOT trip it (that's a real, fast answer), + // and a skip must mark outcome.transient so a pcat-only VIN isn't negative-cached + // for 30m while the circuit self-heals in 30s. + private static readonly PCAT_CB_FAILURE_THRESHOLD = 3; + private static readonly PCAT_CB_COOLDOWN_MS = 30_000; + private static readonly PCAT_CB_COUNTER_TTL_S = 300; + private static readonly PCAT_CB_FAILURE_KEY = "pcat:cb:consec_failures"; + private static readonly PCAT_CB_COOLDOWN_KEY = "pcat:cb:cooldown_until"; + + private async isPcatCircuitOpen(): Promise { + const until = await this.redis.get(VehiclesService.PCAT_CB_COOLDOWN_KEY); + return until !== null && Number(until) > Date.now(); + } + + private async recordPcatFailure(): Promise { + const c = await this.redis.incr(VehiclesService.PCAT_CB_FAILURE_KEY); + await this.redis.expire( + VehiclesService.PCAT_CB_FAILURE_KEY, + VehiclesService.PCAT_CB_COUNTER_TTL_S, + ); + if (c >= VehiclesService.PCAT_CB_FAILURE_THRESHOLD) { + await this.redis.set( + VehiclesService.PCAT_CB_COOLDOWN_KEY, + String(Date.now() + VehiclesService.PCAT_CB_COOLDOWN_MS), + Math.ceil(VehiclesService.PCAT_CB_COOLDOWN_MS / 1000) + 5, + ); + this.logger.warn( + `PCAT circuit opened for ${VehiclesService.PCAT_CB_COOLDOWN_MS}ms (${c} consecutive failures)`, + ); + } + } + + private async recordPcatSuccess(): Promise { + await this.redis.del(VehiclesService.PCAT_CB_FAILURE_KEY); + } + /** * Shared VIN decode chain — Corgi → PartsCatalogs → PL24 → EMEX. * @@ -694,13 +735,43 @@ export class VehiclesService { const pcatStart = Date.now(); let pcatResolved = false; - const pcatPromise = this.partsCatalogsService - .decodeVin(vin, signal, outcome) + // Circuit breaker: when pcat is in a bad transport window, skip it so the + // race isn't dragged; EMEX/PL24 answer instead. A pcat-scoped outcome lets + // the breaker count ONLY pcat's transient faults (not another source's). + const pcatCircuitOpen = await this.isPcatCircuitOpen(); + if (ctx && pcatCircuitOpen) ctx.timings.pcat_circuit_open = true; + const pcatOutcome: { transient: boolean; coldSkip?: boolean } = { transient: false }; + // Fast-fail on a cold JWT pool: EMEX (already racing) or PL24 can answer + // while the pool warms in the background, instead of the user eating the + // multi-second inline capture. The pcat result self-heals on the next decode. + const basePcat: Promise = pcatCircuitOpen + ? (() => { + // Skipped (open) — NOT a definitive miss. Mark transient so a + // pcat-only VIN isn't negative-cached while the circuit self-heals. + this.logger.warn(`PCAT skipped for ${vin}: circuit breaker is open`); + pcatOutcome.transient = true; + return Promise.resolve(null); + })() + : this.partsCatalogsService.decodeVin(vin, signal, pcatOutcome, { blockOnEmpty: false }); + const pcatPromise = basePcat + .then(async (r) => { + // Breaker bookkeeping (skip when already open — nothing was attempted): + // a transient TRANSPORT fault counts toward opening; any definitive + // answer (cars OR a clean "no cars" miss) clears the streak. A deliberate + // cold-pool skip is neutral — it isn't a fault and the pool is warming. + if (!pcatCircuitOpen && !pcatOutcome.coldSkip) { + if (pcatOutcome.transient) await this.recordPcatFailure(); + else await this.recordPcatSuccess(); + } + return r; + }) .catch((err: Error) => { this.logger.warn(`PartsCatalogs decode failed for ${vin}: ${err.message}`); return null; }) .finally(() => { + // Propagate pcat's transient signal to the shared outcome (neg-cache gate). + if (pcatOutcome.transient && outcome) outcome.transient = true; pcatResolved = true; if (ctx && ctx.timings.pcat === undefined) { ctx.timings.pcat = Date.now() - pcatStart;