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

perf(vehicles): dedup, abort budget, circuit breaker for VIN decode

Browse Source 
Production p99 was 18min for failed decodes and 15min for PL24 successes;
same VIN could trigger N parallel 10+ min decodes. Three layered fixes:

- In-flight dedup via Redis SETNX (vin🔒*); concurrent same-VIN requests
  poll the resolve cache instead of re-firing the upstream chain.
- 24h positive cache (vin:resolve:*) and 6h negative cache
  (vin:resolve:neg:*); previously 5min positive / no negative.
- 25s hard abort budget via AbortController; PCAT gets the signal natively
  (AbortSignal.any), PL24/EMEX wrapped with raceWithSignal at the boundary.
  Aborted decodes don't poison the negative cache.
- PCAT/EMEX real race: first definitive single-result wins; the slower
  source is skipped (previously PCAT was always awaited first).
- PL24 circuit breaker: 3 consecutive failures opens a 30s cooldown
  (pl24:cb:cooldown_until); successes reset the counter.
- Stage-level timings in query_logs.timings (jsonb): pcat/emex/pl24/
  lock_wait/cache_hit/aborted. Failed source now logged as "none" or
  "aborted" instead of misleading "corgi".

Verified locally with 3 parallel decodes of a fresh VIN: 1 real decode
(3.59s), 2 lock-waits (3.53s) sharing the result, 4th request 23ms cache
hit. Previously this would have been 3 separate 10+ min PL24 decodes.

Migration 0002 adds query_logs.timings jsonb (NULL default). Must be
applied manually before deploy (deploy.sh does not run db:push).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sase Dev
2026-05-12 11:31:25 +00:00
parent 189d5a9656
commit c4f6cc1caf
9 changed files with 5518 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apps/api/drizzle/0002_boring_the_stranger.sql Normal file
View File

@@ -0,0 +1 @@
ALTER TABLE "query_logs" ADD COLUMN "timings" jsonb;

5173
apps/api/drizzle/meta/0002_snapshot.json Normal file
View File

File diff suppressed because it is too large Load Diff

9
apps/api/drizzle/meta/_journal.json
View File

@@ -15,6 +15,13 @@
"when": 1747039600000,
"tag": "0001_charming_quicksand",
"breakpoints": true
},
{
"idx": 2,
"version": "7",
"when": 1778584574615,
"tag": "0002_boring_the_stranger",
"breakpoints": true
}
]
}
}

1
apps/api/src/admin/admin.service.ts
View File

@@ -256,6 +256,7 @@ export class AdminService {
success: queryLogs.success,
errorMessage: queryLogs.errorMessage,
responseTimeMs: queryLogs.responseTimeMs,
timings: queryLogs.timings,
createdAt: queryLogs.createdAt,
})
.from(queryLogs)

3
apps/api/src/database/schema/core.ts
View File

@@ -203,6 +203,9 @@ export const queryLogs = pgTable(
success: boolean("success").default(true).notNull(),
errorMessage: text("error_message"),
responseTimeMs: integer("response_time_ms"),
// Per-stage durations in ms; e.g. { pcat: 1240, emex: 3000, pl24: 9876, total: 14116, aborted: true }.
// Populated by VehiclesService.resolveVin so admin analytics can attribute slow decodes to the right source.
timings: jsonb("timings"),
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
},
(table) => [

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

@@ -41,9 +41,9 @@ export class PartsCatalogsService {
/**
* VIN decode — returns one or more car matches.
*/
async decodeVin(vin: string): Promise<PcatVinResult | null> {
async decodeVin(vin: string, signal?: AbortSignal): Promise<PcatVinResult | null> {
try {
const data = await this.fetchWithAuth("/car/info", { q: vin });
const data = await this.fetchWithAuth("/car/info", { q: vin }, signal);
if (!data || typeof data !== "object") {
return null;
@@ -165,12 +165,19 @@ export class PartsCatalogsService {
// ─── Private ─────────────────────────────────────────────
private async fetchWithAuth(endpoint: string, params?: Record<string, string>): Promise<any> {
private async fetchWithAuth(
endpoint: string,
params?: Record<string, string>,
externalSignal?: AbortSignal,
): Promise<any> {
const maxRetries = 2;
let session: PcatSession | null = null;
for (let attempt = 0; attempt <= maxRetries; attempt++) {
if (externalSignal?.aborted) {
throw new Error("Request aborted by caller");
}
session = await this.authService.acquireSession();
const url = new URL(`${API_BASE}${endpoint}`);
@@ -181,6 +188,8 @@ export class PartsCatalogsService {
}
try {
const signals = [AbortSignal.timeout(REQUEST_TIMEOUT)];
if (externalSignal) signals.push(externalSignal);
const fetchOptions: RequestInit & { dispatcher?: any } = {
method: "GET",
headers: {
@@ -194,7 +203,7 @@ export class PartsCatalogsService {
"User-Agent":
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36",
},
signal: AbortSignal.timeout(REQUEST_TIMEOUT),
signal: AbortSignal.any(signals),
};
// Use undici ProxyAgent if proxy is configured

6
apps/api/src/redis/redis.service.ts
View File

@@ -28,6 +28,12 @@ export class RedisService implements OnModuleDestroy {
}
}
/** SET key value NX EX ttl — returns true if the key was created. */
async setNx(key: string, value: string, ttlSeconds: number): Promise<boolean> {
const result = await this.client.set(key, value, "EX", ttlSeconds, "NX");
return result === "OK";
}
async setJson(key: string, value: unknown, ttlSeconds?: number): Promise<void> {
await this.set(key, JSON.stringify(value), ttlSeconds);
}

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

@@ -31,6 +31,11 @@ import { PrefetchSource } from "../jobs/prefetch.types";
import { CATALOG_PREFETCH_QUEUE } from "../jobs/queues/catalog-prefetch.queue";
import { RedisService } from "../redis/redis.service";
/** Per-call stage timings, populated by resolveVin so logQuery can persist them. */
interface ResolveContext {
timings: Record<string, number>;
}
interface VinResolveResult {
brandName: string | null;
model: string | null;
@@ -67,6 +72,7 @@ export class VehiclesService {
async decodeVin(vin: string, userId: string, pcatCarId?: string, emexCarIndex?: number) {
const startTime = Date.now();
const ctx: ResolveContext = { timings: {} };
if (!isValidVin(vin)) {
throw new BadRequestException("Geçersiz şase numarası");
@@ -81,33 +87,65 @@ export class VehiclesService {
await this.checkBrandAccess(userId, existing.brandId);
}
await this.ensureUserVehicleLink(userId, existing.id);
await this.logQuery(userId, vin, existing.brandId, "cache", true, Date.now() - startTime);
await this.logQuery(
userId,
vin,
existing.brandId,
"cache",
true,
Date.now() - startTime,
undefined,
{ db_hit: 1 },
);
return existing;
}
// 2. Resolve VIN via cached decode chain (Corgi → PartsCatalogs → PL24 → EMEX)
const resolved = await this.resolveVin(vin, pcatCarId, emexCarIndex, userId);
const resolved = await this.resolveVin(vin, pcatCarId, emexCarIndex, userId, ctx);
if (!resolved) {
const finalSource = ctx.timings.aborted ? "aborted" : "none";
const errMsg = ctx.timings.aborted
? `Decode budget exceeded (${VehiclesService.RESOLVE_BUDGET_MS}ms)`
: "Unknown VIN/brand";
await this.logQuery(
userId,
vin,
null,
"corgi",
finalSource,
false,
Date.now() - startTime,
"Unknown VIN/brand",
errMsg,
ctx.timings,
);
throw new BadRequestException("Şase numarası tanınamadı. Marka desteklenmiyor.");
}
// 2b. If resolveVin returned multiple candidates, return them for frontend selection
if (resolved.pcatCandidates && resolved.pcatCandidates.length > 1) {
await this.logQuery(userId, vin, null, "parts-catalogs", true, Date.now() - startTime);
await this.logQuery(
userId,
vin,
null,
"parts-catalogs",
true,
Date.now() - startTime,
undefined,
ctx.timings,
);
return { candidates: resolved.pcatCandidates, vin, source: "parts-catalogs" };
}
if (resolved.emexCandidates && resolved.emexCandidates.length > 1) {
await this.logQuery(userId, vin, null, "emex", true, Date.now() - startTime);
await this.logQuery(
userId,
vin,
null,
"emex",
true,
Date.now() - startTime,
undefined,
ctx.timings,
);
return { candidates: resolved.emexCandidates, vin, source: "emex" };
}
@@ -169,7 +207,16 @@ export class VehiclesService {
await this.schedulePrefetch(savedVehicle.id, source as PrefetchSource);
}
await this.logQuery(userId, vin, brandId, source, true, Date.now() - startTime);
await this.logQuery(
userId,
vin,
brandId,
source,
true,
Date.now() - startTime,
undefined,
ctx.timings,
);
return savedVehicle;
}
@@ -214,11 +261,67 @@ export class VehiclesService {
};
}
// ─── Resolve cache TTLs & budgets ──────────────────────
// VIN decode result is deterministic; cache aggressively. Negatives shorter
// so transient upstream errors don't poison results for a full day.
private static readonly RESOLVE_TTL_POSITIVE_S = 86_400; // 24h
private static readonly RESOLVE_TTL_NEGATIVE_S = 21_600; // 6h
private static readonly RESOLVE_LOCK_TTL_S = 60;
private static readonly RESOLVE_WAIT_POLL_MS = 250;
private static readonly RESOLVE_WAIT_TIMEOUT_MS = 30_000;
// Hard ceiling for the full decode chain. The previous record was a 17-min
// PL24 decode (responseTimeMs=1006123) that ran long after the request was
// already aborted by TimeoutInterceptor. Anything past this is wasted work.
private static readonly RESOLVE_BUDGET_MS = 25_000;
// ─── PL24 circuit breaker ──────────────────────────────
// PL24 had p95=12min in production. When it goes bad it stays bad — every
// user serially eats the long timeout. After 3 consecutive failures we open
// the circuit for 30s so subsequent decodes fall through to other sources or
// fail fast. The failure counter auto-expires if no recent activity.
private static readonly PL24_CB_FAILURE_THRESHOLD = 3;
private static readonly PL24_CB_COOLDOWN_MS = 30_000;
private static readonly PL24_CB_COUNTER_TTL_S = 300;
private static readonly PL24_CB_FAILURE_KEY = "pl24:cb:consec_failures";
private static readonly PL24_CB_COOLDOWN_KEY = "pl24:cb:cooldown_until";
private async isPl24CircuitOpen(): Promise<boolean> {
const until = await this.redis.get(VehiclesService.PL24_CB_COOLDOWN_KEY);
return until !== null && Number(until) > Date.now();
}
private async recordPl24Failure(): Promise<void> {
const c = await this.redis.incr(VehiclesService.PL24_CB_FAILURE_KEY);
await this.redis.expire(
VehiclesService.PL24_CB_FAILURE_KEY,
VehiclesService.PL24_CB_COUNTER_TTL_S,
);
if (c >= VehiclesService.PL24_CB_FAILURE_THRESHOLD) {
await this.redis.set(
VehiclesService.PL24_CB_COOLDOWN_KEY,
String(Date.now() + VehiclesService.PL24_CB_COOLDOWN_MS),
Math.ceil(VehiclesService.PL24_CB_COOLDOWN_MS / 1000) + 5,
);
this.logger.warn(
`PL24 circuit opened for ${VehiclesService.PL24_CB_COOLDOWN_MS}ms (${c} consecutive failures)`,
);
}
}
private async recordPl24Success(): Promise<void> {
await this.redis.del(VehiclesService.PL24_CB_FAILURE_KEY);
}
/**
* Shared VIN decode chain with 5-minute Redis cache.
* Corgi (offline) → PartsCatalogs → PL24 fallback → EMEX fallback.
* Shared VIN decode chain — Corgi → PartsCatalogs → PL24 → EMEX.
*
* @param pcatCarId If provided, skip resolve chain and use this specific PC car
* Wrapper handles:
* - cache read (positive & negative)
* - in-flight dedup via Redis SETNX lock (so 3 concurrent requests for the
* same VIN don't fire 3 parallel decodes)
* - cache write on completion
*
* @param pcatCarId If provided, skip resolve chain and use this specific PC car
* @param emexCarIndex If provided, skip resolve chain and use this EMEX candidate index
*/
private async resolveVin(
@@ -226,8 +329,9 @@ export class VehiclesService {
pcatCarId?: string,
emexCarIndex?: number,
userId?: string,
ctx?: ResolveContext,
): Promise<VinResolveResult | null> {
// If user selected a specific PC car from candidates, resolve directly
// Candidate-pick paths bypass the shared cache (user-driven, per-request).
if (pcatCarId) {
return this.resolvePcatCarById(vin, pcatCarId);
}
@@ -236,28 +340,148 @@ export class VehiclesService {
}
const cacheKey = `vin:resolve:${vin}`;
const negKey = `vin:resolve:neg:${vin}`;
const lockKey = `vin:lock:${vin}`;
// 1. Cache hits — return immediately.
const cached = await this.redis.getJson<VinResolveResult>(cacheKey);
if (cached) {
this.logger.debug(`VIN resolve cache hit for ${vin}`);
if (ctx) ctx.timings.cache_hit = 1;
return cached;
}
if (await this.redis.exists(negKey)) {
this.logger.debug(`VIN resolve negative cache hit for ${vin}`);
if (ctx) ctx.timings.cache_neg_hit = 1;
return null;
}
const corgiResult = null;
// 2. Try to acquire the in-flight lock.
const acquired = await this.redis.setNx(lockKey, "1", VehiclesService.RESOLVE_LOCK_TTL_S);
if (!acquired) {
// Another request is decoding this VIN — wait for its result.
const waitStart = Date.now();
const waited = await this.waitForResolve(vin, cacheKey, negKey);
if (ctx) ctx.timings.lock_wait = Date.now() - waitStart;
if (waited !== "timeout") return waited;
// Waiter timed out (decode is taking longer than expected). Fall through
// and decode ourselves — duplicate work, but better than blocking forever.
this.logger.warn(`VIN ${vin} lock wait timed out — decoding ourselves`);
}
// 3. Set up a hard abort budget. If the chain exceeds it, every fetch with
// this signal aborts and we return null without poisoning the negative
// cache (transient timeout ≠ unknown VIN).
const ac = new AbortController();
const budgetTimer = setTimeout(() => ac.abort(), VehiclesService.RESOLVE_BUDGET_MS);
try {
const result = await this.doResolveVin(vin, userId, ctx, ac.signal);
if (ac.signal.aborted) {
this.logger.warn(
`VIN ${vin} decode aborted by budget (${VehiclesService.RESOLVE_BUDGET_MS}ms)`,
);
if (ctx) ctx.timings.aborted = 1;
// Don't cache: this was a transient timeout, not a permanent failure.
return null;
}
if (result) {
await this.redis.setJson(cacheKey, result, VehiclesService.RESOLVE_TTL_POSITIVE_S);
} else {
await this.redis.set(negKey, "1", VehiclesService.RESOLVE_TTL_NEGATIVE_S);
}
return result;
} finally {
clearTimeout(budgetTimer);
await this.redis.del(lockKey).catch(() => {
/* lock TTL will expire; ignore */
});
}
}
/**
* Race a promise against an AbortSignal. When the signal aborts, returns null
* even if the underlying promise is still in flight. Use this at integration
* boundaries that don't natively accept AbortSignal.
*/
private raceWithSignal<T>(promise: Promise<T>, signal?: AbortSignal): Promise<T | null> {
if (!signal) return promise;
return Promise.race([
promise,
new Promise<null>((resolve) => {
if (signal.aborted) {
resolve(null);
return;
}
signal.addEventListener("abort", () => resolve(null), { once: true });
}),
]);
}
/** Poll the resolve cache while another request holds the lock. */
private async waitForResolve(
vin: string,
cacheKey: string,
negKey: string,
): Promise<VinResolveResult | null | "timeout"> {
const deadline = Date.now() + VehiclesService.RESOLVE_WAIT_TIMEOUT_MS;
while (Date.now() < deadline) {
await new Promise((r) => setTimeout(r, VehiclesService.RESOLVE_WAIT_POLL_MS));
const cached = await this.redis.getJson<VinResolveResult>(cacheKey);
if (cached) {
this.logger.debug(`VIN ${vin} resolved by concurrent request`);
return cached;
}
if (await this.redis.exists(negKey)) {
this.logger.debug(`VIN ${vin} negative-resolved by concurrent request`);
return null;
}
}
return "timeout";
}
/** Actual decode chain — does NOT touch the cache or lock. */
private async doResolveVin(
vin: string,
userId?: string,
ctx?: ResolveContext,
signal?: AbortSignal,
): Promise<VinResolveResult | null> {
const corgiKnown = false;
let brandName: string | null = null;
// ── Parallel: PartsCatalogs + EMEX (EMEX capped at 3s) ──────────────
const EMEX_RACE_MS = 3000;
const pcatPromise = this.partsCatalogsService.decodeVin(vin).catch((err: Error) => {
this.logger.warn(`PartsCatalogs decode failed for ${vin}: ${err.message}`);
return null;
});
const pcatStart = Date.now();
let pcatResolved = false;
const pcatPromise = this.partsCatalogsService
.decodeVin(vin, signal)
.catch((err: Error) => {
this.logger.warn(`PartsCatalogs decode failed for ${vin}: ${err.message}`);
return null;
})
.finally(() => {
pcatResolved = true;
if (ctx && ctx.timings.pcat === undefined) {
ctx.timings.pcat = Date.now() - pcatStart;
}
});
const emexBasePromise = this.emexService.decodeVinOrCandidates(vin).catch((err: Error) => {
this.logger.warn(`EMEX decode failed for ${vin}: ${err.message}`);
return null;
});
const emexStart = Date.now();
let emexResolved = false;
const emexBasePromise = this.emexService
.decodeVinOrCandidates(vin)
.catch((err: Error) => {
this.logger.warn(`EMEX decode failed for ${vin}: ${err.message}`);
return null;
})
.finally(() => {
emexResolved = true;
if (ctx && ctx.timings.emex === undefined) {
ctx.timings.emex = Date.now() - emexStart;
}
});
// Emex result capped at 3s — if it doesn't arrive in time, falls back to PL24
const emexTimedPromise = Promise.race([
@@ -265,8 +489,36 @@ export class VehiclesService {
new Promise<null>((resolve) => setTimeout(() => resolve(null), EMEX_RACE_MS)),
]);
// Wait for PartsCatalogs first
const pcatResult = await pcatPromise;
// ── Real race: whichever returns a definitive single result first wins ───
// "Definitive" = PCAT has exactly 1 car, OR EMEX returns a non-UNKNOWN single
// vehicle. If the first completer is definitive we skip the rest entirely;
// otherwise we fall back to the previous decision-tree (await both, decide).
const earlyWinner = await Promise.race([
pcatPromise.then((r) => ({ kind: "pcat" as const, r })),
emexTimedPromise.then((r) => ({ kind: "emex" as const, r })),
]);
if (signal?.aborted) return null;
let pcatResult: Awaited<typeof pcatPromise> = null;
let emexResult: Awaited<typeof emexTimedPromise> = null;
if (earlyWinner.kind === "pcat") pcatResult = earlyWinner.r;
else emexResult = earlyWinner.r;
const earlyPcatDefinitive = earlyWinner.kind === "pcat" && earlyWinner.r?.cars?.length === 1;
const earlyEmexDefinitive =
earlyWinner.kind === "emex" &&
earlyWinner.r?.type === "vehicle" &&
earlyWinner.r.vehicle.brand !== "UNKNOWN";
if (!earlyPcatDefinitive && !earlyEmexDefinitive) {
// First completer wasn't definitive — wait for the other one.
if (earlyWinner.kind === "pcat" && !emexResolved) {
emexResult = await emexTimedPromise;
} else if (earlyWinner.kind === "emex" && !pcatResolved) {
pcatResult = await pcatPromise;
}
if (signal?.aborted) return null;
}
// Decision 1: pcat returned exactly 1 car → use it, ignore EMEX
if (pcatResult?.cars?.length === 1) {
@@ -290,12 +542,10 @@ export class VehiclesService {
corgiKnown,
corgiResult: null,
};
await this.redis.setJson(cacheKey, result, 300);
return result;
}
// Decision 2: pcat ≠ 1 → check EMEX result (may already be resolved)
const emexResult = await emexTimedPromise;
// Decision 2: pcat ≠ 1 → check EMEX result (already set by the race above)
const emexSingleVehicle =
emexResult?.type === "vehicle" && emexResult.vehicle.brand !== "UNKNOWN"
? emexResult.vehicle
@@ -316,17 +566,37 @@ export class VehiclesService {
corgiKnown,
corgiResult: null,
};
await this.redis.setJson(cacheKey, result, 300);
return result;
}
// Decision 3: EMEX timed out / returned multiple / returned nothing → PL24
this.logger.log(`PL24 fallback triggered for ${vin} (EMEX: ${emexResult?.type ?? "timeout"})`);
if (this.pl24Service.isDecodeable(vin)) {
const pl24CircuitOpen = await this.isPl24CircuitOpen();
if (pl24CircuitOpen) {
this.logger.warn(`PL24 skipped for ${vin}: circuit breaker is open`);
if (ctx) ctx.timings.pl24_skipped = 1;
}
if (this.pl24Service.isDecodeable(vin) && !pl24CircuitOpen) {
const pl24Start = Date.now();
try {
const pl24Vehicle = await this.pl24Service.decodeVin(vin, userId);
// PL24 internal fetch chain doesn't accept AbortSignal yet; race it at the
// boundary so the budget timer returns control to the caller. Background
// work may continue (PL24 has its own per-fetch 30s cap), but the user
// and HTTP response are unblocked.
const pl24Vehicle = await this.raceWithSignal(
this.pl24Service.decodeVin(vin, userId),
signal,
);
if (ctx) ctx.timings.pl24 = Date.now() - pl24Start;
if (signal?.aborted) {
// Budget aborted while waiting on PL24 — counts as a failure for CB.
await this.recordPl24Failure();
return null;
}
if (pl24Vehicle) {
await this.recordPl24Success();
if (!brandName) brandName = this.pl24Service.getBrandName(vin) || null;
const result: VinResolveResult = {
brandName,
@@ -340,10 +610,11 @@ export class VehiclesService {
corgiKnown,
corgiResult: null,
};
await this.redis.setJson(cacheKey, result, 300);
return result;
}
} catch (err) {
if (ctx) ctx.timings.pl24 = Date.now() - pl24Start;
await this.recordPl24Failure();
this.logger.warn(`PL24 decode failed for ${vin}: ${(err as Error).message}`);
}
}
@@ -707,6 +978,7 @@ export class VehiclesService {
success: boolean,
responseTimeMs: number,
errorMessage?: string,
timings?: Record<string, number>,
) {
try {
await this.db.insert(queryLogs).values({
@@ -717,6 +989,7 @@ export class VehiclesService {
success,
responseTimeMs,
errorMessage: errorMessage || null,
timings: timings && Object.keys(timings).length > 0 ? timings : null,
});
} catch (err) {
this.logger.error("Failed to log query", err);

13
apps/web/src/routes/dashboard/admin/analytics.tsx
View File

@@ -27,6 +27,7 @@ interface QueryLogItem {
success: boolean;
errorMessage: string | null;
responseTimeMs: number | null;
timings: Record<string, number> | null;
createdAt: string;
}
@@ -185,12 +186,20 @@ function AdminAnalyticsPage() {
<div className="text-right">
{log.responseTimeMs !== null ? (
<span
title={
log.timings
? Object.entries(log.timings)
.map(([k, v]) => `${k}: ${v}ms`)
.join("\n")
: undefined
}
className={
log.responseTimeMs > 5000
(log.responseTimeMs > 5000
? "text-red-500"
: log.responseTimeMs > 2000
? "text-amber-500"
: "text-green-500"
: "text-green-500") +
(log.timings ? " cursor-help underline decoration-dotted" : "")
}
>
{log.responseTimeMs}ms