root/sase.tr
1
0
Fork 0
Code Issues 5 Pull Requests 3 Actions 7 Packages Projects Releases Wiki Activity

perf(pcat): rotating Floxy residential transport + resilience (decode latency)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Semih Yesilyurt
2026-06-10 21:19:14 +03:00
parent a237e55874
commit 25ba37103d
4 changed files with 373 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

291
apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts
View File

@@ -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<void> | 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<string, number>();
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<string>("PCAT_USE_PROXY", "true") === "true";
this.proxyHost = this.configService.get<string>("PCAT_PROXY_HOST", DI_HOST);
this.proxyUser = this.configService.get<string>("PCAT_PROXY_USER", DI_DEFAULT_USER);
this.proxyPass = this.configService.get<string>("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<string>("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<string>("PCAT_PROXY_PROVIDER", "floxy").toLowerCase();
const useProxy = cfg.get<string>("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<string>("PCAT_FLOXY_HOST", FLOXY_HOST);
this.floxyPort = toPort("PCAT_FLOXY_PORT", FLOXY_PORT);
this.floxyUser = cfg.get<string>("PCAT_FLOXY_USER", FLOXY_DEFAULT_USER);
this.floxyPass = cfg.get<string>("PCAT_FLOXY_PASS", FLOXY_DEFAULT_PASS);
this.floxyCaptureLifetime = toInt("PCAT_FLOXY_CAPTURE_LIFETIME", FLOXY_CAPTURE_LIFETIME_S);
this.diHost = cfg.get<string>("PCAT_PROXY_HOST", DI_HOST);
this.diUser = cfg.get<string>("PCAT_PROXY_USER", DI_DEFAULT_USER);
this.diPass = cfg.get<string>("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<void> {
@@ -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<PcatSession> {
async acquireSession(opts?: { blockOnEmpty?: boolean }): Promise<PcatSession | null> {
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<void> {
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<boolean> {
private async validateSlot(jwt: PcatJwtToken): Promise<boolean> {
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<PcatJwtToken | null> {
private async attemptCapture(siteUrl: string): Promise<PcatJwtToken | null> {
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<string, unknown> = {};
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");

109
apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts
View File

@@ -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<PcatVinResult | null> {
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<string, string>,
externalSignal?: AbortSignal,
opts?: { timeoutMs?: number; maxRetries?: number },
opts?: { timeoutMs?: number; maxRetries?: number; blockOnEmpty?: boolean },
): Promise<any> {
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;
}

13
apps/api/src/integrations/parts-catalogs/parts-catalogs.types.ts
View File

@@ -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<typeof setTimeout> | 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 {

77
apps/api/src/vehicles/vehicles.service.ts
View File

@@ -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";
@@ -494,6 +494,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<boolean> {
const until = await this.redis.get(VehiclesService.PCAT_CB_COOLDOWN_KEY);
return until !== null && Number(until) > Date.now();
}
private async recordPcatFailure(): Promise<void> {
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<void> {
await this.redis.del(VehiclesService.PCAT_CB_FAILURE_KEY);
}
/**
* Shared VIN decode chain — Corgi → PartsCatalogs → PL24 → EMEX.
*
@@ -700,13 +741,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<PcatVinResult | null> = 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;