feat(proxy): otomatik Floxy→DataImpulse failover (decode regresyonu) #140
@@ -14,6 +14,11 @@
|
||||
import { Injectable, Logger, type OnModuleDestroy, type OnModuleInit } from "@nestjs/common";
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
import { Browser, BrowserContext, Page } from "playwright";
|
||||
import { ProxyHealthService } from "../proxy-telemetry/proxy-health.service";
|
||||
import {
|
||||
classifyTransportError,
|
||||
isProxyConnectFailure,
|
||||
} from "../proxy-telemetry/proxy-telemetry.service";
|
||||
|
||||
const SESSION_TTL_MS = 25 * 60 * 1000; // 25 minutes
|
||||
const MAX_CONCURRENT_PAGES = 3;
|
||||
@@ -76,8 +81,16 @@ export class EmexBrowserService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly floxyPass: string;
|
||||
private readonly floxyLifetime: number;
|
||||
private startedAt = 0;
|
||||
// Provider the live browser was actually launched with. Tracks failover: when
|
||||
// the configured provider is "floxy" but the shared health gate is tripped,
|
||||
// (re)launch picks "dataimpulse" instead. Whole-browser scope because Playwright
|
||||
// binds the proxy at launch — a relaunch is what swaps the exit.
|
||||
private activeProvider: "floxy" | "dataimpulse" | "none" = "none";
|
||||
|
||||
constructor(private configService: ConfigService) {
|
||||
constructor(
|
||||
private configService: ConfigService,
|
||||
private readonly proxyHealth: ProxyHealthService,
|
||||
) {
|
||||
this.semaphore = new Semaphore(MAX_CONCURRENT_PAGES);
|
||||
|
||||
this.useProxy = this.configService.get<string>("EMEX_USE_PROXY", "true") === "true";
|
||||
@@ -206,7 +219,11 @@ export class EmexBrowserService implements OnModuleInit, OnModuleDestroy {
|
||||
const executablePath = process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH;
|
||||
if (executablePath) launchOptions.executablePath = executablePath;
|
||||
|
||||
if (this.proxyProvider === "floxy") {
|
||||
// Pick the provider for THIS launch: honour the configured one, but while the
|
||||
// shared Floxy-health gate is tripped, a "floxy" config launches on DataImpulse.
|
||||
this.activeProvider = this.pickLaunchProvider();
|
||||
|
||||
if (this.activeProvider === "floxy") {
|
||||
// Sticky Floxy session for the browser's lifetime → one residential exit IP
|
||||
// for the whole chained scrape; a fresh id is minted on each (re)launch.
|
||||
const sid = Math.random().toString(36).slice(2, 10);
|
||||
@@ -222,14 +239,18 @@ export class EmexBrowserService implements OnModuleInit, OnModuleDestroy {
|
||||
this.logger.log(
|
||||
`Using Floxy residential proxy: ${this.floxyHost}:${this.floxyPort} (sticky ${this.floxyLifetime}s)`,
|
||||
);
|
||||
} else if (this.proxyProvider === "dataimpulse") {
|
||||
} else if (this.activeProvider === "dataimpulse") {
|
||||
const port = this.randomProxyPort();
|
||||
launchOptions.proxy = {
|
||||
server: `http://${this.proxyHost}:${port}`,
|
||||
username: this.proxyUsername,
|
||||
password: this.proxyPassword,
|
||||
};
|
||||
this.logger.log(`Using DataImpulse proxy: ${this.proxyHost}:${port}`);
|
||||
this.logger.log(
|
||||
`Using DataImpulse proxy: ${this.proxyHost}:${port}${
|
||||
this.proxyProvider === "floxy" ? " (Floxy failover)" : ""
|
||||
}`,
|
||||
);
|
||||
}
|
||||
|
||||
this.browser = await chromium.launch(launchOptions);
|
||||
@@ -270,34 +291,94 @@ export class EmexBrowserService implements OnModuleInit, OnModuleDestroy {
|
||||
await this.launchBrowser();
|
||||
}
|
||||
|
||||
/** Provider for the next launch: DataImpulse while the Floxy gate is tripped. */
|
||||
private pickLaunchProvider(): "floxy" | "dataimpulse" | "none" {
|
||||
if (this.proxyProvider === "floxy" && this.proxyHealth.isFloxyDown()) {
|
||||
return "dataimpulse"; // DI creds are always present (env defaults) when proxy on
|
||||
}
|
||||
return this.proxyProvider;
|
||||
}
|
||||
|
||||
private async relaunchBrowser(): Promise<void> {
|
||||
await this.closeBrowser();
|
||||
await this.launchBrowser();
|
||||
}
|
||||
|
||||
/**
|
||||
* Visit baseUrl to establish/refresh ASP.NET session cookie
|
||||
* Trip Floxy failover from a scrape-level transport failure (a connect error
|
||||
* that surfaced AFTER the session was established, so ensureSession's own gate
|
||||
* didn't catch it). Opens the shared cooldown and forces the next acquirePage
|
||||
* to re-establish the session — which switches the browser to DataImpulse.
|
||||
*/
|
||||
async tripFloxyFailover(reason?: string): Promise<void> {
|
||||
this.proxyHealth.reportFloxyFailure(reason);
|
||||
if (this.activeProvider === "floxy") {
|
||||
this.sessionExpiry = 0; // force ensureSession → provider switch on next acquire
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Visit baseUrl to establish/refresh ASP.NET session cookie. Doubles as the
|
||||
* Floxy health gate for the browser path: only at a session boundary, switch
|
||||
* the browser's proxy to match pickLaunchProvider() (fail over to DataImpulse
|
||||
* when Floxy is down, probe Floxy again once the gate clears). A connect
|
||||
* failure during the probe trips the gate and retries once on DataImpulse.
|
||||
*/
|
||||
private async ensureSession(): Promise<void> {
|
||||
// Only re-evaluate the provider when establishing a fresh session, so an
|
||||
// in-flight scrape on a working browser is never interrupted mid-flow.
|
||||
if (Date.now() >= this.sessionExpiry && this.proxyProvider !== "none") {
|
||||
const desired = this.pickLaunchProvider();
|
||||
if (this.browser?.isConnected() && desired !== this.activeProvider) {
|
||||
this.logger.log(`EMEX switching browser proxy: ${this.activeProvider} → ${desired}`);
|
||||
await this.relaunchBrowser();
|
||||
}
|
||||
}
|
||||
|
||||
if (Date.now() < this.sessionExpiry) return;
|
||||
|
||||
this.logger.log("Establishing EMEX session...");
|
||||
if (!this.context) throw new Error("EMEX browser context not initialized");
|
||||
const ctx = this.context;
|
||||
const page = await ctx.newPage();
|
||||
try {
|
||||
await page.goto(EMEX_BASE_URL, {
|
||||
waitUntil: "networkidle",
|
||||
timeout: 30000,
|
||||
});
|
||||
// Up to 2 attempts: a Floxy tunnel failure trips the gate, relaunches on
|
||||
// DataImpulse, and tries once more so the caller still gets a live session.
|
||||
for (let attempt = 1; attempt <= 2; attempt++) {
|
||||
if (!this.context) throw new Error("EMEX browser context not initialized");
|
||||
const ctx = this.context;
|
||||
const page = await ctx.newPage();
|
||||
try {
|
||||
this.logger.log(`Establishing EMEX session (${this.activeProvider})...`);
|
||||
await page.goto(EMEX_BASE_URL, {
|
||||
waitUntil: "networkidle",
|
||||
timeout: 30000,
|
||||
});
|
||||
|
||||
const cookies = await ctx.cookies();
|
||||
const session = cookies.find((c) => c.name === "ASP.NET_SessionId");
|
||||
if (session) {
|
||||
this.sessionExpiry = Date.now() + SESSION_TTL_MS;
|
||||
this.logger.log("Session established, TTL 25 min");
|
||||
} else {
|
||||
this.logger.warn("No session cookie found after visiting baseUrl");
|
||||
// Still set a short TTL to avoid hammering
|
||||
this.sessionExpiry = Date.now() + 60_000;
|
||||
if (this.activeProvider === "floxy") this.proxyHealth.reportFloxySuccess();
|
||||
|
||||
const cookies = await ctx.cookies();
|
||||
const session = cookies.find((c) => c.name === "ASP.NET_SessionId");
|
||||
if (session) {
|
||||
this.sessionExpiry = Date.now() + SESSION_TTL_MS;
|
||||
this.logger.log("Session established, TTL 25 min");
|
||||
} else {
|
||||
this.logger.warn("No session cookie found after visiting baseUrl");
|
||||
// Still set a short TTL to avoid hammering
|
||||
this.sessionExpiry = Date.now() + 60_000;
|
||||
}
|
||||
return;
|
||||
} catch (err) {
|
||||
await page.close().catch(() => {});
|
||||
const canFailover =
|
||||
attempt < 2 &&
|
||||
this.proxyProvider === "floxy" &&
|
||||
this.activeProvider === "floxy" &&
|
||||
isProxyConnectFailure(err);
|
||||
if (canFailover) {
|
||||
this.proxyHealth.reportFloxyFailure(`emex session: ${classifyTransportError(err)}`);
|
||||
await this.relaunchBrowser(); // gate now tripped → relaunch picks DataImpulse
|
||||
continue;
|
||||
}
|
||||
throw err;
|
||||
} finally {
|
||||
await page.close().catch(() => {});
|
||||
}
|
||||
} finally {
|
||||
await page.close();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -19,6 +19,11 @@ const browser = {} as never;
|
||||
const redis = { set: vi.fn() } as never;
|
||||
const posthog = {} as never;
|
||||
const proxyTelemetry = { record: vi.fn() } as never;
|
||||
const proxyHealth = {
|
||||
isFloxyDown: vi.fn(() => false),
|
||||
reportFloxyFailure: vi.fn(),
|
||||
reportFloxySuccess: vi.fn(),
|
||||
} as never;
|
||||
|
||||
describe("EmexService proxy-port coercion (regression)", () => {
|
||||
it("does not throw 'Invalid URL' when ports arrive as strings over a real range", () => {
|
||||
@@ -29,7 +34,9 @@ describe("EmexService proxy-port coercion (regression)", () => {
|
||||
EMEX_PROXY_PORT_END: "10099",
|
||||
});
|
||||
// Pre-fix: threw "Invalid URL" here (port "4510001" > 65535).
|
||||
expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
|
||||
expect(
|
||||
() => new EmexService(config, browser, redis, posthog, proxyTelemetry, proxyHealth),
|
||||
).not.toThrow();
|
||||
});
|
||||
|
||||
it("constructs with a single-port range (string env)", () => {
|
||||
@@ -38,7 +45,9 @@ describe("EmexService proxy-port coercion (regression)", () => {
|
||||
EMEX_PROXY_PORT_START: "823",
|
||||
EMEX_PROXY_PORT_END: "823",
|
||||
});
|
||||
expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
|
||||
expect(
|
||||
() => new EmexService(config, browser, redis, posthog, proxyTelemetry, proxyHealth),
|
||||
).not.toThrow();
|
||||
});
|
||||
|
||||
it("falls back to a valid default when the port env is garbage", () => {
|
||||
@@ -47,11 +56,15 @@ describe("EmexService proxy-port coercion (regression)", () => {
|
||||
EMEX_PROXY_PORT_START: "not-a-number",
|
||||
EMEX_PROXY_PORT_END: "999999",
|
||||
});
|
||||
expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
|
||||
expect(
|
||||
() => new EmexService(config, browser, redis, posthog, proxyTelemetry, proxyHealth),
|
||||
).not.toThrow();
|
||||
});
|
||||
|
||||
it("constructs cleanly with the proxy disabled", () => {
|
||||
const config = makeConfig({ EMEX_USE_PROXY: "false" });
|
||||
expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
|
||||
expect(
|
||||
() => new EmexService(config, browser, redis, posthog, proxyTelemetry, proxyHealth),
|
||||
).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -23,9 +23,11 @@ import { ProxyAgent } from "undici";
|
||||
import { isBackfillContext } from "../../jobs/prefetch-context";
|
||||
import { PostHogService } from "../../posthog/posthog.service";
|
||||
import { RedisService } from "../../redis/redis.service";
|
||||
import { ProxyHealthService } from "../proxy-telemetry/proxy-health.service";
|
||||
import {
|
||||
ProxyTelemetryService,
|
||||
classifyTransportError,
|
||||
isProxyConnectFailure,
|
||||
} from "../proxy-telemetry/proxy-telemetry.service";
|
||||
import { parseUnitLeaves, parseVehicleTree } from "./emex-tree.parser";
|
||||
import { EmexBrowserService } from "./emex.browser";
|
||||
@@ -147,6 +149,7 @@ export class EmexService {
|
||||
private redis: RedisService,
|
||||
private posthog: PostHogService,
|
||||
private proxyTelemetry: ProxyTelemetryService,
|
||||
private proxyHealth: ProxyHealthService,
|
||||
) {
|
||||
// __dirname is apps/api/src/integrations/emex/ or dist/integrations/emex/
|
||||
// Scraper lives at <monorepo-root>/scripts/emex-vin-scraper.js
|
||||
@@ -383,8 +386,14 @@ export class EmexService {
|
||||
// and a different proxy IP must not "retry" it away.
|
||||
const schedule: Array<"dataimpulse" | "floxy"> = [];
|
||||
if (this.emexProxyProvider === "floxy") {
|
||||
if (this.emexFloxy) schedule.push("floxy", "floxy");
|
||||
if (this.emexProxy) schedule.push("dataimpulse");
|
||||
if (this.emexFloxy && this.proxyHealth.isFloxyDown() && this.emexProxy) {
|
||||
// Floxy gate tripped → DataImpulse-first (rotating ports cover its ~50%
|
||||
// dead ports), with one Floxy probe last so a recovery is still detected.
|
||||
schedule.push("dataimpulse", "dataimpulse", "floxy");
|
||||
} else {
|
||||
if (this.emexFloxy) schedule.push("floxy", "floxy");
|
||||
if (this.emexProxy) schedule.push("dataimpulse");
|
||||
}
|
||||
} else if (this.emexProxyProvider === "dataimpulse") {
|
||||
if (this.emexProxy) schedule.push("dataimpulse", "dataimpulse", "dataimpulse");
|
||||
if (this.emexFloxy) schedule.push("floxy", "floxy");
|
||||
@@ -417,6 +426,7 @@ export class EmexService {
|
||||
throw new Error(`EMEX HTTP ${res.status} for ${url}`);
|
||||
}
|
||||
if (provider === "floxy") {
|
||||
this.proxyHealth.reportFloxySuccess(); // Floxy answered — clear cooldown
|
||||
this.logger.log(`EMEX fetch via Floxy fallback succeeded for ${url}`);
|
||||
}
|
||||
return await res.text();
|
||||
@@ -446,6 +456,11 @@ export class EmexService {
|
||||
/fetch failed|ECONNRESET|ECONNREFUSED|EAI_AGAIN|socket hang up|other side closed|terminated|UND_ERR/i.test(
|
||||
`${e.message} ${String(e.cause ?? "")}`,
|
||||
));
|
||||
// A Floxy connect failure (dead exit / tunnel refused) trips the shared
|
||||
// gate so EMEX + pcat fail over to DataImpulse for the cooldown window.
|
||||
if (provider === "floxy" && isProxyConnectFailure(e)) {
|
||||
this.proxyHealth.reportFloxyFailure(`emex http: ${classifyTransportError(e)}`);
|
||||
}
|
||||
if (!transient || attempt === maxAttempts) break;
|
||||
// A Floxy transport failure means the current sticky exit IP is dead —
|
||||
// roll to a fresh session so the next Floxy attempt gets a new IP.
|
||||
@@ -909,6 +924,12 @@ export class EmexService {
|
||||
} catch (error) {
|
||||
const err = error as Error;
|
||||
|
||||
// Floxy tunnel dead mid-scrape → trip failover so the next acquirePage
|
||||
// re-establishes the session on DataImpulse.
|
||||
if (isProxyConnectFailure(err)) {
|
||||
void this.browserService.tripFloxyFailover(`emex decode: ${classifyTransportError(err)}`);
|
||||
}
|
||||
|
||||
if (
|
||||
err instanceof BadRequestException ||
|
||||
err instanceof ServiceUnavailableException ||
|
||||
@@ -1052,6 +1073,10 @@ export class EmexService {
|
||||
return { parts: [], schemaImageUrl: null, hotspots: [], schemaWidth: 0, schemaHeight: 0 };
|
||||
} catch (error) {
|
||||
const err = error as Error;
|
||||
// Floxy tunnel dead mid-scrape → trip failover for the next acquirePage.
|
||||
if (isProxyConnectFailure(err)) {
|
||||
void this.browserService.tripFloxyFailover(`emex parts: ${classifyTransportError(err)}`);
|
||||
}
|
||||
this.logger.error(`Failed to fetch category parts: ${err.message}`);
|
||||
return { parts: [], schemaImageUrl: null, hotspots: [], schemaWidth: 0, schemaHeight: 0 };
|
||||
} finally {
|
||||
|
||||
@@ -23,9 +23,11 @@ import { Injectable, Logger, type OnModuleDestroy, type OnModuleInit } from "@ne
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
import { Browser, BrowserContext } from "playwright";
|
||||
import { RedisService } from "../../redis/redis.service";
|
||||
import { ProxyHealthService } from "../proxy-telemetry/proxy-health.service";
|
||||
import {
|
||||
ProxyTelemetryService,
|
||||
classifyTransportError,
|
||||
isProxyConnectFailure,
|
||||
resolveExitIp,
|
||||
} from "../proxy-telemetry/proxy-telemetry.service";
|
||||
import { JwtSlot, PcatJwtToken, PcatSession } from "./parts-catalogs.types";
|
||||
@@ -190,6 +192,7 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
private configService: ConfigService,
|
||||
private readonly redis: RedisService,
|
||||
private readonly proxyTelemetry: ProxyTelemetryService,
|
||||
private readonly proxyHealth: ProxyHealthService,
|
||||
) {
|
||||
const cfg = this.configService;
|
||||
|
||||
@@ -254,17 +257,29 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
// Telemetry identity: sticky Floxy session id or DataImpulse port. Null on
|
||||
// rotating legs (exit IP changes per request — nothing stable to attribute).
|
||||
sessionKey: string | null;
|
||||
// The provider actually used for this leg (may differ from the configured
|
||||
// provider when Floxy is in failover). Drives accurate telemetry + which
|
||||
// provider's health to report success/failure against.
|
||||
provider: ProxyProvider;
|
||||
} {
|
||||
if (this.proxyProvider === "none")
|
||||
return { proxyUrl: null, proxyConfig: null, sessionKey: null };
|
||||
return { proxyUrl: null, proxyConfig: null, sessionKey: null, provider: "none" };
|
||||
|
||||
if (this.proxyProvider === "dataimpulse") {
|
||||
// Floxy-primary but the shared health gate is tripped (Floxy down — balance
|
||||
// or tunnel) → fail over to the DataImpulse datacenter pool for this leg so
|
||||
// captures/calls keep landing instead of dying on a dead residential proxy.
|
||||
const useDataImpulse =
|
||||
this.proxyProvider === "dataimpulse" ||
|
||||
(this.proxyProvider === "floxy" && this.proxyHealth.isFloxyDown());
|
||||
|
||||
if (useDataImpulse) {
|
||||
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 },
|
||||
sessionKey: `di:${port}`,
|
||||
provider: "dataimpulse",
|
||||
};
|
||||
}
|
||||
|
||||
@@ -281,6 +296,7 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
proxyUrl: `http://${this.floxyUser}:${password}@${this.floxyHost}:${this.floxyPort}`,
|
||||
proxyConfig: { server, username: this.floxyUser, password },
|
||||
sessionKey,
|
||||
provider: "floxy",
|
||||
};
|
||||
}
|
||||
|
||||
@@ -737,7 +753,7 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
|
||||
// Sticky capture proxy: one residential exit IP pinned for the lifetime of
|
||||
// this capture so the partner site + widget see a coherent session.
|
||||
const { proxyConfig, proxyUrl, sessionKey } = this.buildProxy("capture");
|
||||
const { proxyConfig, proxyUrl, sessionKey, provider } = this.buildProxy("capture");
|
||||
|
||||
// The session is sticky, so a parallel ipify probe exits from the SAME IP
|
||||
// the capture will use — that's what makes banned-IP tracking concrete.
|
||||
@@ -747,7 +763,7 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
void exitIpPromise.then((exitIp) =>
|
||||
this.proxyTelemetry.record({
|
||||
service: "pcat_capture",
|
||||
provider: this.proxyProvider,
|
||||
provider,
|
||||
sessionKey,
|
||||
exitIp,
|
||||
targetHost: new URL(siteUrl).hostname,
|
||||
@@ -834,6 +850,12 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
// constants above.
|
||||
gotoOk = false;
|
||||
this.logger.debug(`Navigation ended: ${(navErr as Error).message?.slice(0, 80)}`);
|
||||
// A tunnel/connect failure here (not a plain slow-site timeout) means the
|
||||
// Floxy exit is dead — trip the shared gate so this + EMEX fail over to
|
||||
// DataImpulse for the cooldown window.
|
||||
if (provider === "floxy" && isProxyConnectFailure(navErr)) {
|
||||
this.proxyHealth.reportFloxyFailure(`pcat capture: ${classifyTransportError(navErr)}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Poll for token — short cap after healthy goto, brief grace after fail
|
||||
@@ -847,6 +869,8 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
|
||||
if (capturedToken) {
|
||||
logCapture(true);
|
||||
// Floxy answered — clear any failover cooldown so we resume residential.
|
||||
if (provider === "floxy") this.proxyHealth.reportFloxySuccess();
|
||||
this.logger.log(`Token captured in ${elapsed}ms from ${new URL(siteUrl).hostname}`);
|
||||
return capturedToken;
|
||||
}
|
||||
@@ -858,6 +882,9 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
return null;
|
||||
} catch (err) {
|
||||
logCapture(false, classifyTransportError(err));
|
||||
if (provider === "floxy" && isProxyConnectFailure(err)) {
|
||||
this.proxyHealth.reportFloxyFailure(`pcat capture: ${classifyTransportError(err)}`);
|
||||
}
|
||||
this.logger.warn(`Token capture error: ${(err as Error).message}`);
|
||||
return null;
|
||||
} finally {
|
||||
|
||||
@@ -11,10 +11,12 @@ import { Injectable, Logger } from "@nestjs/common";
|
||||
import { isBackfillContext } from "../../jobs/prefetch-context";
|
||||
import { PostHogService } from "../../posthog/posthog.service";
|
||||
import { RedisService } from "../../redis/redis.service";
|
||||
import { ProxyHealthService } from "../proxy-telemetry/proxy-health.service";
|
||||
import {
|
||||
ProxyTelemetryService,
|
||||
classifyTransportError,
|
||||
describeProxyUrl,
|
||||
isProxyConnectFailure,
|
||||
} from "../proxy-telemetry/proxy-telemetry.service";
|
||||
import { PartsCatalogsAuthService } from "./parts-catalogs-auth.service";
|
||||
import {
|
||||
@@ -68,6 +70,7 @@ export class PartsCatalogsService {
|
||||
private redis: RedisService,
|
||||
private posthog: PostHogService,
|
||||
private proxyTelemetry: ProxyTelemetryService,
|
||||
private proxyHealth: ProxyHealthService,
|
||||
) {}
|
||||
|
||||
/** Mark parts-catalogs as actively used (5min TTL) to defer prefetch worker */
|
||||
@@ -324,6 +327,10 @@ export class PartsCatalogsService {
|
||||
if (response.ok) {
|
||||
// Token proven good on this IP — clear any accrued failure streak.
|
||||
session._slot.failCount = 0;
|
||||
// Floxy answered at transport level — clear any failover cooldown.
|
||||
if (describeProxyUrl(session.proxyUrl).provider === "floxy") {
|
||||
this.proxyHealth.reportFloxySuccess();
|
||||
}
|
||||
return await response.json();
|
||||
}
|
||||
|
||||
@@ -352,6 +359,11 @@ export class PartsCatalogsService {
|
||||
success: false,
|
||||
durationMs: Date.now() - attemptStart,
|
||||
});
|
||||
// Floxy exit refused/closed the CONNECT → trip the shared gate so the
|
||||
// next capture/call (and EMEX) fails over to DataImpulse.
|
||||
if (describeProxyUrl(session.proxyUrl).provider === "floxy" && isProxyConnectFailure(e)) {
|
||||
this.proxyHealth.reportFloxyFailure(`pcat call: ${classifyTransportError(e)}`);
|
||||
}
|
||||
}
|
||||
// Retry transient TRANSPORT failures only: request timeouts and undici
|
||||
// network errors ("TypeError: fetch failed" — a dropped/reset DataImpulse
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
import type { ConfigService } from "@nestjs/config";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { ProxyHealthService } from "./proxy-health.service";
|
||||
import { isProxyConnectFailure } from "./proxy-telemetry.service";
|
||||
|
||||
function makeConfig(overrides: Record<string, unknown> = {}): ConfigService {
|
||||
return {
|
||||
get: vi.fn((key: string, def?: unknown) =>
|
||||
Object.hasOwn(overrides, key) ? overrides[key] : def,
|
||||
),
|
||||
} as unknown as ConfigService;
|
||||
}
|
||||
|
||||
describe("ProxyHealthService — Floxy failover gate", () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
});
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("starts healthy (Floxy not down)", () => {
|
||||
const h = new ProxyHealthService(makeConfig());
|
||||
expect(h.isFloxyDown()).toBe(false);
|
||||
});
|
||||
|
||||
it("trips for the cooldown window then self-heals when it lapses", () => {
|
||||
const h = new ProxyHealthService(makeConfig({ FLOXY_FAILOVER_COOLDOWN_MS: 180_000 }));
|
||||
h.reportFloxyFailure("test");
|
||||
expect(h.isFloxyDown()).toBe(true);
|
||||
expect(h.cooldownRemainingSec()).toBe(180);
|
||||
|
||||
vi.advanceTimersByTime(179_000);
|
||||
expect(h.isFloxyDown()).toBe(true);
|
||||
|
||||
vi.advanceTimersByTime(2_000); // past the 180s window
|
||||
expect(h.isFloxyDown()).toBe(false);
|
||||
});
|
||||
|
||||
it("a Floxy success clears the cooldown immediately", () => {
|
||||
const h = new ProxyHealthService(makeConfig({ FLOXY_FAILOVER_COOLDOWN_MS: 180_000 }));
|
||||
h.reportFloxyFailure();
|
||||
expect(h.isFloxyDown()).toBe(true);
|
||||
h.reportFloxySuccess();
|
||||
expect(h.isFloxyDown()).toBe(false);
|
||||
});
|
||||
|
||||
it("re-reporting a failure extends the window", () => {
|
||||
const h = new ProxyHealthService(makeConfig({ FLOXY_FAILOVER_COOLDOWN_MS: 100_000 }));
|
||||
h.reportFloxyFailure();
|
||||
vi.advanceTimersByTime(90_000);
|
||||
h.reportFloxyFailure(); // extends from now
|
||||
vi.advanceTimersByTime(20_000); // 110s since first failure, 20s since second
|
||||
expect(h.isFloxyDown()).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("isProxyConnectFailure", () => {
|
||||
it("matches proxy/tunnel connection failures (the Floxy-down signal)", () => {
|
||||
for (const msg of [
|
||||
"page.goto: net::ERR_TUNNEL_CONNECTION_FAILED at https://emexdwc.ae/",
|
||||
"net::ERR_PROXY_CONNECTION_FAILED",
|
||||
"TypeError: fetch failed",
|
||||
"connect ECONNREFUSED 1.2.3.4:12321",
|
||||
"Client network socket disconnected (ECONNRESET)",
|
||||
]) {
|
||||
expect(isProxyConnectFailure(new Error(msg)), msg).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
it("does NOT match a plain slow-site navigation timeout", () => {
|
||||
const e = new Error("Timeout 30000ms exceeded.");
|
||||
e.name = "TimeoutError";
|
||||
expect(isProxyConnectFailure(e)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,68 @@
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
|
||||
/**
|
||||
* Shared Floxy-health gate. Floxy is the residential primary for EMEX + pcat
|
||||
* egress (HTTP fetch, Playwright capture, browser scrape). When Floxy goes dark
|
||||
* — balance exhausted (proxy returns 402) or the entry node refuses CONNECT
|
||||
* (`ERR_TUNNEL_CONNECTION_FAILED` / `fetch failed`) — every consumer otherwise
|
||||
* keeps hammering the dead proxy and the decode success rate collapses (this
|
||||
* happened 2026-06-11/13: ~17k tunnel failures/24h, decode 78% → 45%).
|
||||
*
|
||||
* This singleton is the one place that answers "should we route through Floxy
|
||||
* right now?". Any consumer that sees a Floxy transport failure trips the gate;
|
||||
* for the cooldown window all consumers fail over to the DataImpulse datacenter
|
||||
* pool instead. The gate self-heals: once the cooldown lapses, the next request
|
||||
* probes Floxy again and `reportFloxySuccess()` clears it (or re-trips if still
|
||||
* down). A live Floxy answer at any time clears the cooldown immediately.
|
||||
*
|
||||
* In-memory only and deliberately dependency-light (no DB, no Redis): a single
|
||||
* api/worker process is the egress unit, and a stale cooldown is self-correcting
|
||||
* within one window. Cooldown is tunable via FLOXY_FAILOVER_COOLDOWN_MS.
|
||||
*/
|
||||
@Injectable()
|
||||
export class ProxyHealthService {
|
||||
private readonly logger = new Logger(ProxyHealthService.name);
|
||||
private readonly cooldownMs: number;
|
||||
private floxyDownUntil = 0;
|
||||
|
||||
constructor(private readonly config: ConfigService) {
|
||||
const raw = Number(this.config.get("FLOXY_FAILOVER_COOLDOWN_MS", 180_000));
|
||||
this.cooldownMs = Number.isFinite(raw) && raw >= 0 ? raw : 180_000;
|
||||
}
|
||||
|
||||
/** True while Floxy is in cooldown — consumers should route via DataImpulse. */
|
||||
isFloxyDown(): boolean {
|
||||
return Date.now() < this.floxyDownUntil;
|
||||
}
|
||||
|
||||
/**
|
||||
* Floxy transport failure observed — open (or extend) the cooldown so every
|
||||
* consumer fails over to DataImpulse. Logged once per outage, not per failure,
|
||||
* so a 6k-failure storm doesn't flood the logs.
|
||||
*/
|
||||
reportFloxyFailure(reason?: string): void {
|
||||
const wasDown = this.isFloxyDown();
|
||||
this.floxyDownUntil = Date.now() + this.cooldownMs;
|
||||
if (!wasDown) {
|
||||
this.logger.warn(
|
||||
`Floxy proxy unhealthy${reason ? ` (${reason})` : ""} — failing over to DataImpulse for ${Math.round(
|
||||
this.cooldownMs / 1000,
|
||||
)}s`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/** A Floxy request just succeeded — Floxy is healthy again, clear the gate. */
|
||||
reportFloxySuccess(): void {
|
||||
if (this.isFloxyDown()) {
|
||||
this.logger.log("Floxy proxy recovered — resuming Floxy-primary routing");
|
||||
}
|
||||
this.floxyDownUntil = 0;
|
||||
}
|
||||
|
||||
/** Seconds remaining in the current cooldown (0 when healthy). */
|
||||
cooldownRemainingSec(): number {
|
||||
return Math.max(0, Math.ceil((this.floxyDownUntil - Date.now()) / 1000));
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,12 @@
|
||||
import { Module } from "@nestjs/common";
|
||||
import { ProxyHealthService } from "./proxy-health.service";
|
||||
import { ProxyTelemetryService } from "./proxy-telemetry.service";
|
||||
|
||||
// Imported by PartsCatalogsModule and EmexModule; Nest module caching makes the
|
||||
// service a singleton (one shared flush buffer) across both.
|
||||
// services singletons (one shared flush buffer / one shared Floxy-health gate)
|
||||
// across both.
|
||||
@Module({
|
||||
providers: [ProxyTelemetryService],
|
||||
exports: [ProxyTelemetryService],
|
||||
providers: [ProxyTelemetryService, ProxyHealthService],
|
||||
exports: [ProxyTelemetryService, ProxyHealthService],
|
||||
})
|
||||
export class ProxyTelemetryModule {}
|
||||
|
||||
@@ -52,6 +52,25 @@ export function classifyTransportError(err: unknown): string {
|
||||
return "transport";
|
||||
}
|
||||
|
||||
/**
|
||||
* True when an error means we could not even reach the upstream THROUGH the
|
||||
* proxy — the proxy entry node refused/closed/reset the CONNECT, or undici/
|
||||
* Chromium reported a tunnel/proxy connection failure ("fetch failed",
|
||||
* ERR_TUNNEL_CONNECTION_FAILED, ERR_PROXY_CONNECTION_FAILED, ECONNREFUSED…).
|
||||
* This is the "the proxy itself is dead" signal that should trip Floxy failover.
|
||||
*
|
||||
* Deliberately EXCLUDES a plain navigation/request TimeoutError with no proxy
|
||||
* marker: a slow-but-alive residential exit shouldn't trip failover — only a
|
||||
* genuine connection failure should.
|
||||
*/
|
||||
export function isProxyConnectFailure(err: unknown): boolean {
|
||||
const e = err as Error & { cause?: unknown };
|
||||
const text = `${e?.name ?? ""} ${e?.message ?? ""} ${String(e?.cause ?? "")}`;
|
||||
return /ERR_TUNNEL_CONNECTION_FAILED|ERR_PROXY_CONNECTION_FAILED|ERR_SOCKS_CONNECTION_FAILED|NS_ERROR_PROXY|ECONNREFUSED|ECONNRESET|UND_ERR_SOCKET|UND_ERR_CONNECT_TIMEOUT|socket hang up|other side closed|tunnel|proxy connection|fetch failed/i.test(
|
||||
text,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Derive provider + session key from a proxy URL the way the integrations
|
||||
* build them: Floxy passwords may embed `_session-<id>_lifetime-<n>` (sticky);
|
||||
@@ -79,10 +98,7 @@ export function describeProxyUrl(proxyUrl: string | null | undefined): {
|
||||
* probe through the same proxy. Only meaningful for sticky sessions — on a
|
||||
* rotating leg the probe and the real request exit from different IPs.
|
||||
*/
|
||||
export async function resolveExitIp(
|
||||
proxyUrl: string,
|
||||
timeoutMs = 5_000,
|
||||
): Promise<string | null> {
|
||||
export async function resolveExitIp(proxyUrl: string, timeoutMs = 5_000): Promise<string | null> {
|
||||
try {
|
||||
const { ProxyAgent } = await import("undici");
|
||||
const res = await fetch("https://api.ipify.org?format=text", {
|
||||
@@ -138,7 +154,9 @@ export class ProxyTelemetryService implements OnModuleDestroy {
|
||||
);
|
||||
} catch (err) {
|
||||
// Telemetry must never matter more than the product. Drop the batch.
|
||||
this.logger.debug(`proxy_logs flush failed (${batch.length} rows): ${(err as Error).message}`);
|
||||
this.logger.debug(
|
||||
`proxy_logs flush failed (${batch.length} rows): ${(err as Error).message}`,
|
||||
);
|
||||
} finally {
|
||||
this.flushing = false;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user