diff --git a/apps/api/src/jobs/prefetch-utils.ts b/apps/api/src/jobs/prefetch-utils.ts index 45bfbcb..9332035 100644 --- a/apps/api/src/jobs/prefetch-utils.ts +++ b/apps/api/src/jobs/prefetch-utils.ts @@ -74,14 +74,21 @@ export async function checkCooldown(redis: RedisService, source: string): Promis } } -/** Current hour (0–23) in Europe/Istanbul. */ -function currentIstanbulHour(): number { +/** Europe/Istanbul hour (0–23) at an arbitrary instant. */ +function istanbulHourAt(ms: number): number { const hourStr = new Intl.DateTimeFormat("en-US", { timeZone: "Europe/Istanbul", hour: "numeric", hour12: false, - }).format(new Date()); - return Number.parseInt(hourStr, 10); + }).format(new Date(ms)); + // `% 24` because the h24 hour cycle renders midnight as "24", which would put + // the hour outside every window and silently park the source forever. + return Number.parseInt(hourStr, 10) % 24; +} + +/** Current hour (0–23) in Europe/Istanbul. */ +function currentIstanbulHour(): number { + return istanbulHourAt(Date.now()); } /** @@ -115,6 +122,35 @@ export function checkTimeWindow(source: string): void { } } +/** + * The first instant at or after `fromMs` that falls inside `source`'s scrape + * window. Returns `fromMs` unchanged when the window is disabled (the default + * 0–24) or when `fromMs` is already inside it. + * + * WHY (plv2.md, finding consumers_jobs-03 — the budget/window deadlock): + * `checkSourceDailyBudget` defers a spent source to the next UTC midnight. With + * PREFETCH_PL24_START=9 that midnight lands at 03:00 Europe/Istanbul — six hours + * BEFORE the window opens. The woken job therefore did no work, threw + * `time-window`, and was deferred again to 09:00 — by which point the fresh + * daily budget had already been spent by the same stampede of no-op wake-ups. + * Measured on prod 2026-09-20: 600/600 pl24 budget consumed, 0 catalog requests + * and 0 new categories for the whole day. Landing the deferral inside the window + * breaks the cycle. + */ +export function alignToWindow(source: string, fromMs: number): number { + if (source !== "pl24") return fromMs; + if (PL24_WINDOW_START <= 0 && PL24_WINDOW_END >= 24) return fromMs; + let t = fromMs; + // Step by the hour rather than constructing a local-midnight date: DST-safe and + // free of month/year rollover edge cases. 48 steps covers any window shape. + for (let i = 0; i < 48; i++) { + const h = istanbulHourAt(t); + if (h >= PL24_WINDOW_START && h < PL24_WINDOW_END) return t; + t += 3_600_000; + } + return t; +} + /** * Milliseconds until the next 09:00 Europe/Istanbul. */ diff --git a/apps/api/src/jobs/prefetch-window-budget.spec.ts b/apps/api/src/jobs/prefetch-window-budget.spec.ts new file mode 100644 index 0000000..44b82e6 --- /dev/null +++ b/apps/api/src/jobs/prefetch-window-budget.spec.ts @@ -0,0 +1,283 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +/** + * Regression lock for the PL24 budget/window DEADLOCK (plv2.md, consumers_jobs-03). + * + * Two bugs combined to kill PL24 prefetch outright on prod: + * 1. `checkSourceDailyBudget` was debited in `process()` while the + * business-hours gate still sat at the top of each handler — so a job that + * woke outside the window paid a budget unit to do nothing. + * 2. A spent source was deferred to the next UTC midnight, which is 03:00 + * Europe/Istanbul — six hours BEFORE a 09:00 window opens. Every deferred + * job therefore woke early, burned a unit of the fresh daily budget, hit + * the window gate and was deferred again. + * + * Measured on prod 2026-09-20 before the fix: `prefetch:daily:pl24` at 600/600 + * with ZERO catalog requests and ZERO new pl24 categories for the whole day. + * + * The window constants are read at module load, so every case imports the + * modules fresh with the env already set. + */ + +const REAL_ENV = { ...process.env }; + +/** 2026-09-20 00:00 UTC = 03:00 Europe/Istanbul — outside a 09:00–18:00 window. */ +const OUTSIDE_MS = Date.UTC(2026, 8, 20, 0, 0, 0); +/** 2026-09-20 09:00 UTC = 12:00 Europe/Istanbul — inside it. */ +const INSIDE_MS = Date.UTC(2026, 8, 20, 9, 0, 0); + +function istanbulHour(ms: number): number { + return ( + Number.parseInt( + new Intl.DateTimeFormat("en-US", { + timeZone: "Europe/Istanbul", + hour: "numeric", + hour12: false, + }).format(new Date(ms)), + 10, + ) % 24 + ); +} + +async function loadModules(env: Record) { + vi.resetModules(); + for (const [k, v] of Object.entries(env)) { + if (v === undefined) delete process.env[k]; + else process.env[k] = v; + } + const utils = await import("./prefetch-utils"); + const worker = await import("./prefetch-worker.service"); + return { utils, worker }; +} + +/** Minimal deps: only what `process()` touches before dispatching a job. */ +function makeDeps(redisOverrides: Record = {}) { + const incrCalls: string[] = []; + const redis = { + exists: vi.fn(async (..._a: unknown[]) => false), + get: vi.fn(async (..._a: unknown[]): Promise => null), + set: vi.fn(async (..._a: unknown[]) => undefined), + del: vi.fn(async (..._a: unknown[]) => undefined), + incr: vi.fn(async (k: string) => { + incrCalls.push(k); + return 1; + }), + expire: vi.fn(async (..._a: unknown[]) => undefined), + ttl: vi.fn(async (..._a: unknown[]) => -2), // no cooldown key + setNx: vi.fn(async (..._a: unknown[]) => true), + getJson: vi.fn(async (..._a: unknown[]): Promise => null), + setJson: vi.fn(async (..._a: unknown[]) => undefined), + ...redisOverrides, + }; + const queue = { + name: "catalog-prefetch", + add: vi.fn(async (..._a: unknown[]) => undefined), + getJob: vi.fn(async (..._a: unknown[]): Promise => null), + }; + const categoriesService = { + getCategoryWithParts: vi.fn(async (..._a: unknown[]) => ({ parts: [] })), + getChildren: vi.fn(async (..._a: unknown[]) => []), + }; + return { redis, queue, categoriesService, incrCalls }; +} + +function makeJob(name: string, data: Record) { + return { + name, + data, + moveToDelayed: vi.fn(async (..._a: unknown[]) => undefined), + attemptsMade: 0, + }; +} + +const dailyIncrs = (keys: string[]) => keys.filter((k) => k.startsWith("prefetch:daily:pl24")); + +describe("alignToWindow", () => { + afterEach(() => { + process.env = { ...REAL_ENV }; + }); + + it("pushes a pre-window instant into the configured window", async () => { + const { utils } = await loadModules({ PREFETCH_PL24_START: "9", PREFETCH_PL24_END: "18" }); + const aligned = utils.alignToWindow("pl24", OUTSIDE_MS); + expect(aligned).toBeGreaterThan(OUTSIDE_MS); + const h = istanbulHour(aligned); + expect(h).toBeGreaterThanOrEqual(9); + expect(h).toBeLessThan(18); + }); + + it("leaves an in-window instant untouched", async () => { + const { utils } = await loadModules({ PREFETCH_PL24_START: "9", PREFETCH_PL24_END: "18" }); + expect(utils.alignToWindow("pl24", INSIDE_MS)).toBe(INSIDE_MS); + }); + + it("is a no-op for sources that have no window", async () => { + const { utils } = await loadModules({ PREFETCH_PL24_START: "9", PREFETCH_PL24_END: "18" }); + expect(utils.alignToWindow("emex", OUTSIDE_MS)).toBe(OUTSIDE_MS); + expect(utils.alignToWindow("parts-catalogs", OUTSIDE_MS)).toBe(OUTSIDE_MS); + }); + + it("is a no-op when the window is disabled (the default 0–24)", async () => { + const { utils } = await loadModules({ + PREFETCH_PL24_START: undefined, + PREFETCH_PL24_END: undefined, + }); + expect(utils.alignToWindow("pl24", OUTSIDE_MS)).toBe(OUTSIDE_MS); + }); +}); + +describe("process() gate order — window before daily budget", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + process.env = { ...REAL_ENV }; + }); + + it("does NOT debit the daily budget for a job deferred by the window", async () => { + const { worker } = await loadModules({ + PREFETCH_PL24_START: "9", + PREFETCH_PL24_END: "18", + PL24_TR_DISABLED: undefined, + }); + vi.setSystemTime(OUTSIDE_MS); + const { redis, queue, categoriesService, incrCalls } = makeDeps(); + const svc = new worker.PrefetchWorkerService( + queue as never, + queue as never, + categoriesService as never, + redis as never, + { payload: vi.fn(async () => ({})) } as never, + {} as never, + ); + const job = makeJob("prefetch-parts", { + vehicleId: "v1", + categoryId: "c1", + source: "pl24", + fast: true, + }); + + await expect( + (svc as never as { process: (j: unknown, t?: string) => Promise }).process(job, "tok"), + ).rejects.toThrow(); + + expect(job.moveToDelayed).toHaveBeenCalled(); + expect(dailyIncrs(incrCalls)).toHaveLength(0); + // …and the handler never ran, so nothing was fetched upstream either. + expect(categoriesService.getCategoryWithParts).not.toHaveBeenCalled(); + }); + + it("defers the window-blocked job to an instant inside the window", async () => { + const { worker } = await loadModules({ + PREFETCH_PL24_START: "9", + PREFETCH_PL24_END: "18", + PL24_TR_DISABLED: undefined, + }); + vi.setSystemTime(OUTSIDE_MS); + const { redis, queue, categoriesService } = makeDeps(); + const svc = new worker.PrefetchWorkerService( + queue as never, + queue as never, + categoriesService as never, + redis as never, + { payload: vi.fn(async () => ({})) } as never, + {} as never, + ); + const job = makeJob("prefetch-parts", { + vehicleId: "v1", + categoryId: "c1", + source: "pl24", + fast: true, + }); + await expect( + (svc as never as { process: (j: unknown, t?: string) => Promise }).process(job, "tok"), + ).rejects.toThrow(); + + const [when] = job.moveToDelayed.mock.calls[0] as [number]; + const h = istanbulHour(when); + expect(h).toBeGreaterThanOrEqual(9); + expect(h).toBeLessThan(18); + }); + + it("debits the daily budget once the window is open", async () => { + const { worker } = await loadModules({ + PREFETCH_PL24_START: "9", + PREFETCH_PL24_END: "18", + PL24_TR_DISABLED: undefined, + }); + vi.setSystemTime(INSIDE_MS); + const { redis, queue, categoriesService, incrCalls } = makeDeps(); + const svc = new worker.PrefetchWorkerService( + queue as never, + queue as never, + categoriesService as never, + redis as never, + { payload: vi.fn(async () => ({})) } as never, + {} as never, + ); + const job = makeJob("prefetch-parts", { + vehicleId: "v1", + categoryId: "c1", + source: "pl24", + fast: true, + }); + + await (svc as never as { process: (j: unknown, t?: string) => Promise }).process( + job, + "tok", + ); + + expect(dailyIncrs(incrCalls)).toHaveLength(1); + expect(categoriesService.getCategoryWithParts).toHaveBeenCalledWith("c1"); + }); +}); + +describe("checkSourceDailyBudget — spent source retries inside the window", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + process.env = { ...REAL_ENV }; + }); + + it("never parks a spent source at 03:00 Istanbul again", async () => { + const { worker } = await loadModules({ + PREFETCH_PL24_START: "9", + PREFETCH_PL24_END: "18", + PREFETCH_DAILY_PL24: "600", + }); + vi.setSystemTime(INSIDE_MS); + // Counter already at the fast-lane ceiling → the next job must be deferred. + const { redis, queue, categoriesService } = makeDeps({ + get: vi.fn(async (k: string) => (String(k).startsWith("prefetch:daily:pl24") ? "600" : null)), + }); + const svc = new worker.PrefetchWorkerService( + queue as never, + queue as never, + categoriesService as never, + redis as never, + { payload: vi.fn(async () => ({})) } as never, + {} as never, + ); + const job = makeJob("prefetch-parts", { + vehicleId: "v1", + categoryId: "c1", + source: "pl24", + fast: true, + }); + + await expect( + (svc as never as { process: (j: unknown, t?: string) => Promise }).process(job, "tok"), + ).rejects.toThrow(); + + const [when] = job.moveToDelayed.mock.calls[0] as [number]; + // Past the UTC rollover… + expect(when).toBeGreaterThan(Date.UTC(2026, 8, 21, 0, 0, 0)); + // …and inside the window, not at 03:00 Istanbul like the old deferral. + const h = istanbulHour(when); + expect(h).toBeGreaterThanOrEqual(9); + expect(h).toBeLessThan(18); + }); +}); diff --git a/apps/api/src/jobs/prefetch-worker.service.ts b/apps/api/src/jobs/prefetch-worker.service.ts index 4e61ca3..fb854e8 100644 --- a/apps/api/src/jobs/prefetch-worker.service.ts +++ b/apps/api/src/jobs/prefetch-worker.service.ts @@ -17,6 +17,7 @@ import { QUEUE_NAMES, getBullConnection } from "./bull.config"; import { backfillContext } from "./prefetch-context"; import { RateLimitError, + alignToWindow, checkCooldown, checkTimeWindow, initProgress, @@ -355,6 +356,17 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy { ) { const lane = (job.data as { fast?: boolean }).fast ? "fast" : "main"; await this.checkSourceRate(data.source, lane); + // Cooldown + business-hours window BEFORE the daily charge. These two + // gates used to live at the top of each handler, i.e. AFTER the budget + // was already debited, so every job that woke outside the window paid a + // budget unit to do nothing. Combined with a deferral target of "next UTC + // midnight" (= 03:00 Europe/Istanbul, six hours before a 09:00 window + // opens) that formed a closed loop: the whole daily allowance was burned + // by no-op wake-ups before the window ever opened, so the source never + // ran again. Measured on prod 2026-09-20: pl24 at 600/600 with 0 catalog + // requests and 0 new categories for the day. Order matters here. + await checkCooldown(this.redis, data.source); + checkTimeWindow(data.source); // Daily budget AFTER the per-minute gate: a job deferred on the minute // ceiling above never reaches here, so rate-limited retries don't inflate // the daily counter — only jobs about to do real work are counted. The @@ -421,8 +433,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy { const { vehicleId, source, fast = false } = job.data; this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`); - await checkCooldown(this.redis, source); - checkTimeWindow(source); + // Cooldown + time-window are enforced in process() before the daily budget + // is debited — see the comment there; re-checking here would be a no-op. // Already flagged as poison (tree exceeded CATEGORY_CAP on a prior run) — skip. if (await this.redis.exists(this.poisonKey(vehicleId))) { @@ -563,8 +575,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy { const { vehicleId, categoryId, source, depth, fast = false } = job.data; this.logger.log(`[prefetch] Children for category=${categoryId}, depth=${depth}`); - await checkCooldown(this.redis, source); - checkTimeWindow(source); + // Cooldown + time-window are enforced in process() before the daily budget + // is debited — see the comment there; re-checking here would be a no-op. const depthCeiling = maxDepthFor(source, fast); if (depth >= depthCeiling) { @@ -631,8 +643,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy { const { vehicleId, categoryId, source } = job.data; this.logger.log(`[prefetch] Parts for category=${categoryId}`); - await checkCooldown(this.redis, source); - checkTimeWindow(source); + // Cooldown + time-window are enforced in process() before the daily budget + // is debited — see the comment there; re-checking here would be a no-op. try { await this.categoriesService.getCategoryWithParts(categoryId); @@ -1214,11 +1226,17 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy { // deferred job wakes in the SAME millisecond (observed: 11495 jobs all at // 00:00:01 UTC) — the promotion lands as one burst and the pressure signal // flaps. Other sources keep flowing (per-job defer, not a worker pause). - const msLeft = dayMs - (now % dayMs) + 1000 + Math.floor(Math.random() * 45 * 60_000); + const rollover = now + dayMs - (now % dayMs) + 1000 + Math.floor(Math.random() * 45 * 60_000); + // Land the retry INSIDE the source's scrape window. The UTC rollover alone + // is 03:00 Europe/Istanbul, so with a 09:00 window every deferred job woke + // six hours early, failed the window check and was deferred again — the + // other half of the deadlock fixed in process(). alignToWindow is a no-op + // when no window is configured (the default). + const msLeft = Math.max(1000, alignToWindow(source, rollover) - now); if (n === limit) { this.logger.warn( `[prefetch] ${source} daily budget hit (lane=${lane}, ${n}/${limit} of ${max}) — ` + - `deferring ~${Math.round(msLeft / 3_600_000)}h until the window rolls`, + `deferring ~${Math.round(msLeft / 3_600_000)}h to the next in-window slot`, ); } throw new RateLimitError(msLeft, "source-rate");