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

Merge pull request 'fix(backfill): in-process scan timer (cron marker buried behind 444k backlog)' (#136) from fix/backfill-inprocess-scan into main

Browse Source
This commit was merged in pull request #136.
This commit is contained in:
root
2026-06-12 09:14:25 +00:00
parent a304420456 7d10107f4b
commit 6a54e63c0f
3 changed files with 97 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
apps/api/src/jobs/jobs.module.ts
View File

@@ -1,7 +1,7 @@
import { Inject, Module, type OnModuleDestroy, type OnModuleInit } from "@nestjs/common";
import { Queue } from "bullmq";
import { CategoriesModule } from "../categories/categories.module";
import { isCatalogBackfillEnabled, isLifecycleEmailEnabled } from "./prefetch-utils";
import { isLifecycleEmailEnabled } from "./prefetch-utils";
import { PrefetchWorkerService } from "./prefetch-worker.service";
import {
CATALOG_PREFETCH_QUEUE,
@@ -104,36 +104,16 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
console.log("[jobs] Skipped lifecycle-email cron (not prod host)");
}
// Catalog backfill: every hour at :00. Scans for decoded vehicles whose catalog
// isn't fully prefetched and queues them. Self-throttled (queue-depth guard),
// cooldown- and business-hours-aware — handled by PrefetchWorkerService.
// PRODUCTION (sase.tr) ONLY: dev.sase.tr is a separate DB and must not
// sweep/scrape. NODE_ENV is "production" on BOTH, so gate on the prod host.
if (isCatalogBackfillEnabled()) {
await this.catalogPrefetchQueue.upsertJobScheduler(
"catalog-backfill-hourly",
{ pattern: "0 * * * *" },
{
name: "backfill-scan",
data: {},
opts: {
removeOnComplete: { count: 48 },
removeOnFail: { count: 100 },
// lifo so the scan job lands at the tail of the wait list and runs on
// the next tick instead of queueing behind the deep-drill backlog —
// otherwise the hourly scan is buried for days and never fires. BullMQ
// drains wait (RPOPLPUSH from the tail) before the prioritized ZSET, so
// lifo — not `priority` — is what jumps an already-deep wait queue.
lifo: true,
},
},
);
console.log("[jobs] Registered catalog-backfill cron: 0 * * * *");
} else {
// Clean up any stale scheduler (e.g. if NODE_ENV changed) and stay idle.
await this.catalogPrefetchQueue.removeJobScheduler("catalog-backfill-hourly").catch(() => {});
console.log("[jobs] Skipped catalog-backfill cron (not prod host)");
}
// Catalog backfill now runs on an IN-PROCESS hourly timer inside
// PrefetchWorkerService — NOT a BullMQ cron. A cron marker is a delayed job that
// BullMQ promotes to the wait-list HEAD with LPUSH (lifo ignored), and the worker
// pops from the tail, so behind a deep prefetch backlog the scan was buried for
// days and never fired. Remove any previously-registered cron scheduler here so
// stale markers stop being produced (the in-process timer self-gates on the prod
// host). The "backfill-scan" job handler is kept so any already-queued legacy
// marker still runs harmlessly when reached.
await this.catalogPrefetchQueue.removeJobScheduler("catalog-backfill-hourly").catch(() => {});
console.log("[jobs] catalog-backfill cron removed (scan runs in-process)");
// Parça Uzmanları sezon kapanışı: her ayın 1'i 00:00 Türkiye saati —
// biten ayın ilk 3'ünü admin@sase.tr'ye mailler (üyelik günlerini admin

51
apps/api/src/jobs/prefetch-worker.service.spec.ts
View File

@@ -27,6 +27,7 @@ function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
exists: vi.fn(async () => false), // no source cooldown, no in-flight guard
get: vi.fn(async () => null), // no no-result residue
set: vi.fn(async () => undefined),
setNx: vi.fn(async () => true), // scan lock acquired by default
getJson: vi.fn(async () => null), // Phase-2 cursor empty
setJson: vi.fn(async () => undefined),
};
@@ -54,8 +55,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
describe("enqueueInit", () => {
it("sets lifo + data.fast when fast", async () => {
const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
await (service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> })
.enqueueInit("v1", "emex", true);
await (
service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> }
).enqueueInit("v1", "emex", true);
const [name, data, jobOpts] = queue.add.mock.calls[0];
expect(name).toBe("prefetch-init");
expect(data).toMatchObject({ vehicleId: "v1", source: "emex", fast: true });
@@ -64,8 +66,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
it("omits lifo when not fast", async () => {
const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
await (service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> })
.enqueueInit("v1", "emex", false);
await (
service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> }
).enqueueInit("v1", "emex", false);
const [, data, jobOpts] = queue.add.mock.calls[0];
expect(data).toMatchObject({ fast: false });
expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
@@ -77,7 +80,14 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
await (service as never as { addJob: (n: string, d: unknown) => Promise<void> }).addJob(
"prefetch-children",
{ vehicleId: "v1", categoryId: "c1", source: "emex", action: "children", depth: 1, fast: true },
{
vehicleId: "v1",
categoryId: "c1",
source: "emex",
action: "children",
depth: 1,
fast: true,
},
);
const [, , jobOpts] = queue.add.mock.calls[0];
expect(jobOpts).toMatchObject({ lifo: true });
@@ -102,7 +112,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
waiting: 50_000,
limitResults: [[{ id: "empty1", source: "emex" }]], // Phase-1 query result
});
await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
await (
service as never as { processBackfillScan: () => Promise<void> }
).processBackfillScan();
expect(queue.add).toHaveBeenCalledTimes(1);
const [name, data, jobOpts] = queue.add.mock.calls[0];
@@ -123,7 +135,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
[{ id: "stale1", source: "emex", createdAt: new Date("2026-01-01T00:00:00Z") }], // Phase-2
],
});
await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
await (
service as never as { processBackfillScan: () => Promise<void> }
).processBackfillScan();
expect(queue.add).toHaveBeenCalledTimes(1);
const [, data, jobOpts] = queue.add.mock.calls[0];
@@ -136,7 +150,28 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
it("skips entirely when the backfill gate is off (not prod host)", async () => {
process.env.CATALOG_BACKFILL_ENABLED = "false";
const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
await (
service as never as { processBackfillScan: () => Promise<void> }
).processBackfillScan();
expect(queue.add).not.toHaveBeenCalled();
});
});
describe("runBackfillScan (in-process timer entry, NX-locked)", () => {
it("runs the scan when the lock is acquired", async () => {
const { service, queue, redis } = makeDeps({
waiting: 50_000,
limitResults: [[{ id: "empty1", source: "emex" }]],
});
await (service as never as { runBackfillScan: () => Promise<void> }).runBackfillScan();
expect(redis.setNx).toHaveBeenCalledWith("prefetch:backfill:lock", "1", expect.any(Number));
expect(queue.add).toHaveBeenCalledTimes(1); // Phase-1 init enqueued
});
it("skips the scan when the lock is already held", async () => {
const { service, queue, redis } = makeDeps({ waiting: 50_000, limitResults: [] });
redis.setNx.mockResolvedValueOnce(false);
await (service as never as { runBackfillScan: () => Promise<void> }).runBackfillScan();
expect(queue.add).not.toHaveBeenCalled();
});
});

43
apps/api/src/jobs/prefetch-worker.service.ts
View File

@@ -45,6 +45,14 @@ const BACKFILL_SOURCES = ["pl24", "emex", "parts-catalogs"];
/** Redis key holding the rolling rescan cursor (last createdAt seen). */
const BACKFILL_CURSOR_KEY = "prefetch:backfill:cursor";
// ── In-process scan timer (replaces the BullMQ cron, which gets buried) ──
/** How often the in-process backfill scan runs. */
const BACKFILL_SCAN_INTERVAL_MS = 60 * 60 * 1000;
/** Delay before the first scan after boot, so startup settles first. */
const BACKFILL_SCAN_KICKOFF_MS = 60 * 1000;
/** NX-lock TTL (seconds) — just under the interval so each cycle re-acquires. */
const BACKFILL_SCAN_LOCK_TTL_S = 55 * 60;
// ── Throughput knobs (env-tunable so prod can ramp without a redeploy) ──
/**
* Concurrent jobs the worker runs. emex/pl24 fetches parallelise well, so >1
@@ -75,6 +83,8 @@ const NORESULT_TTL_S = (Number(process.env.PREFETCH_NORESULT_TTL_DAYS) || 7) * 8
export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(PrefetchWorkerService.name);
private worker: Worker | null = null;
private backfillKickoff?: ReturnType<typeof setTimeout>;
private backfillInterval?: ReturnType<typeof setInterval>;
constructor(
@Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
@@ -110,15 +120,48 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
this.logger.log(
`[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, ${WORKER_RATE_MAX} jobs/min, pcatPace=${PCAT_PACE_MS}ms)`,
);
// Hourly backfill scan — run IN-PROCESS, not as a BullMQ cron job. A cron
// scheduler enqueues a delayed marker; BullMQ promotes delayed jobs to the
// wait-list HEAD with LPUSH (lifo is ignored on delayed promotion) while the
// worker pops from the tail, so behind a deep prefetch backlog the scan marker
// is buried for days and never fires. An in-process timer sidesteps the queue
// entirely; the scan's Phase-1 then enqueues `lifo` init jobs that DO jump the
// wait list. Prod-host gated (same as the scan body); single-fired via a Redis
// NX lock so restarts / multiple instances don't double-scan.
if (isCatalogBackfillEnabled()) {
const run = () =>
backfillContext.run(true, () =>
this.runBackfillScan().catch((err) =>
this.logger.error(`[backfill] scan crashed: ${(err as Error).message}`),
),
);
this.backfillKickoff = setTimeout(run, BACKFILL_SCAN_KICKOFF_MS);
this.backfillInterval = setInterval(run, BACKFILL_SCAN_INTERVAL_MS);
this.logger.log("[backfill] in-process hourly scan enabled");
}
}
async onModuleDestroy() {
if (this.backfillKickoff) clearTimeout(this.backfillKickoff);
if (this.backfillInterval) clearInterval(this.backfillInterval);
if (this.worker) {
await this.worker.close();
this.worker = null;
}
}
/**
* Hourly scan entry point used by the in-process timer. A Redis NX lock keeps it
* single-fire across restarts and (future) multiple instances; processBackfillScan
* is idempotent, so a missed cycle is harmless.
*/
private async runBackfillScan(): Promise<void> {
const got = await this.redis.setNx("prefetch:backfill:lock", "1", BACKFILL_SCAN_LOCK_TTL_S);
if (!got) return;
await this.processBackfillScan();
}
private async process(job: Job, token?: string): Promise<void> {
try {
const data = job.data as { source?: string };