diff --git a/.env.example b/.env.example index 7844b1b..e725937 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,20 @@ DATABASE_URL=postgresql://dxd:dxd@localhost:5432/scraper # Redis REDIS_URL=redis://localhost:6379 +# Worker recovery +# BullMQ lock duration. Defaults to 10 minutes. +# WORKER_LOCK_DURATION_MS=600000 +# Grace period before orphaned active crawls are considered stale. Defaults to 10 minutes. +# ORPHAN_CRAWL_GRACE_MS=600000 +# Reconcile interval for orphaned active crawls (in milliseconds). Defaults to 2 minutes. +# This determines how frequently the system checks for orphaned crawls. +# Should be less than ORPHAN_CRAWL_GRACE_MS to detect stale crawls before they expire. +# Works in conjunction with WORKER_LOCK_DURATION_MS to manage worker recovery timing. +# ORPHAN_CRAWL_RECONCILE_INTERVAL_MS=120000 +# Temporary escape hatch to restore the old skip-lock-renewal behavior. +# Accepts true/1/yes/on to enable. +# WORKER_SKIP_LOCK_RENEWAL=false + # Storage STORAGE_TYPE=auto LOCAL_STORAGE_PATH=./data @@ -29,4 +43,4 @@ FRONTEND_URL=http://localhost:5173 # CORS_ALLOWED_ORIGINS= # Web -VITE_API_URL=http://localhost:3001 +VITE_API_URL=http://localhost:3001 \ No newline at end of file diff --git a/package.json b/package.json index 037dd39..26029a4 100644 --- a/package.json +++ b/package.json @@ -17,7 +17,7 @@ "start:web": "bun run --filter @dxd/web start", "start:worker": "bun run --filter @dxd/worker start", "lint": "turbo lint", - "test": "bun test apps/api/src apps/web/src/lib packages/scraper/src", + "test": "bun test apps/api/src apps/web/src/lib packages/scraper/src services/worker/src", "verify": "bun run lint && bun run test", "db:generate": "turbo db:generate --filter=@dxd/api", "db:migrate": "turbo db:migrate --filter=@dxd/api", diff --git a/packages/scraper/src/asset-downloader.test.ts b/packages/scraper/src/asset-downloader.test.ts index bc1c5ff..cf9ccbc 100644 --- a/packages/scraper/src/asset-downloader.test.ts +++ b/packages/scraper/src/asset-downloader.test.ts @@ -85,4 +85,77 @@ describe("AssetDownloader", () => { assert.equal(result, "https://example.com/hero.png"); assert.ok(logs.some((message) => message.includes("Skipping image asset https://example.com/hero.png"))); }); + + it("truncates oversized asset basenames before writing to disk", async () => { + const outputDir = await fs.mkdtemp(path.join(os.tmpdir(), "dxd-asset-long-name-")); + createdTempDirs.push(outputDir); + + const longBaseName = "hero-".repeat(80); + globalThis.fetch = async () => + new Response(new Uint8Array([1, 2, 3]), { + status: 200, + headers: { + "content-type": "image/webp", + "content-length": "3", + }, + }); + + const assets = new AssetDownloader(outputDir); + await assets.init(); + + const localPath = await assets.downloadAsset(`https://example.com/images/${longBaseName}.webp`, "image"); + const filename = path.basename(localPath); + + assert.ok(filename.length <= 200, `expected filename <= 200 chars, got ${filename.length}`); + assert.match(filename, /-[a-f0-9]{10}\.webp$/); + + const saved = await fs.readFile(path.join(outputDir, localPath.slice(1))); + assert.deepEqual(Array.from(saved), [1, 2, 3]); + }); + + it("keeps generated filenames within the cap when the source extension is pathologically long", async () => { + const outputDir = await fs.mkdtemp(path.join(os.tmpdir(), "dxd-asset-long-ext-")); + createdTempDirs.push(outputDir); + + const longExt = `.${"x".repeat(220)}`; + globalThis.fetch = async () => + new Response("body { color: red; }", { + status: 200, + headers: { + "content-type": "text/css", + "content-length": "20", + }, + }); + + const assets = new AssetDownloader(outputDir); + await assets.init(); + + const localPath = await assets.downloadAsset(`https://example.com/css/site${longExt}`, "css"); + const filename = path.basename(localPath); + + assert.ok(filename.length <= 200, `expected filename <= 200 chars, got ${filename.length}`); + + const saved = await fs.readFile(path.join(outputDir, localPath.slice(1)), "utf8"); + assert.equal(saved, "body { color: red; }"); + }); + + it("preserves chunk filenames exactly", async () => { + const outputDir = await fs.mkdtemp(path.join(os.tmpdir(), "dxd-asset-chunk-name-")); + createdTempDirs.push(outputDir); + + globalThis.fetch = async () => + new Response("console.log('chunk');", { + status: 200, + headers: { + "content-type": "application/javascript", + "content-length": "21", + }, + }); + + const assets = new AssetDownloader(outputDir); + await assets.init(); + + const localPath = await assets.downloadAsset("https://example.com/js/runtime.achunk.abcdef123456.js", "js"); + assert.equal(localPath, "/js/runtime.achunk.abcdef123456.js"); + }); }); diff --git a/packages/scraper/src/asset-downloader.ts b/packages/scraper/src/asset-downloader.ts index cf479f4..15ccb2e 100644 --- a/packages/scraper/src/asset-downloader.ts +++ b/packages/scraper/src/asset-downloader.ts @@ -361,7 +361,7 @@ export class AssetDownloader { } else { // For other assets, use slugified name with hash for deduplication const hash = crypto.createHash("sha1").update(assetUrl).digest("hex").slice(0, 10); - filename = `${slugify(baseName)}-${hash}${safeExt}`; + filename = buildBoundedAssetFilename(baseName, hash, safeExt); } const relativeDir = path.relative(this.outputDir, targetDir) || ""; @@ -951,6 +951,18 @@ function inferCategoryFromExt(url: string): AssetCategory | undefined { return undefined; } +const MAX_GENERATED_ASSET_FILENAME_LENGTH = 200; + +function buildBoundedAssetFilename(baseName: string, hash: string, ext: string): string { + const maxExtLength = Math.max(0, MAX_GENERATED_ASSET_FILENAME_LENGTH - 1 - hash.length - 1); + const boundedExt = ext.slice(0, maxExtLength); + const slug = slugify(baseName); + const reservedLength = 1 + hash.length + boundedExt.length; + const maxBaseLength = Math.max(1, MAX_GENERATED_ASSET_FILENAME_LENGTH - reservedLength); + const boundedBase = slug.slice(0, maxBaseLength).replace(/-+$/g, "") || "asset"; + return `${boundedBase}-${hash}${boundedExt}`; +} + function slugify(value: string): string { const normalized = value .toLowerCase() diff --git a/services/worker/src/processor.ts b/services/worker/src/processor.ts index 119d7aa..fdaf4f7 100644 --- a/services/worker/src/processor.ts +++ b/services/worker/src/processor.ts @@ -18,6 +18,7 @@ import path from "node:path"; import archiver from "archiver"; import { once } from "node:events"; import { captureMemorySnapshot, formatMemorySnapshot, type MemorySnapshot } from "./memory.js"; +import { getWorkerRuntimeConfig } from "./worker-config.js"; const redisUrl = process.env.REDIS_URL || "redis://localhost:6379"; const storage = getStorage(); @@ -990,8 +991,7 @@ async function processCrawlJob(job: Job) { }); } -async function reconcileOrphanedCrawls(): Promise { - const orphanGraceMs = parsePositiveIntEnv("ORPHAN_CRAWL_GRACE_MS", 5 * 60 * 1000); +async function reconcileOrphanedCrawls(orphanGraceMs = getWorkerRuntimeConfig().orphanGraceMs): Promise { const cutoff = new Date(Date.now() - orphanGraceMs); const possiblyOrphaned = await db.query.crawls.findMany({ @@ -1137,51 +1137,44 @@ function attachWorkerLogging(worker: Worker, label: string, reconcileTimer } export function startWorker() { - const workerConcurrency = parsePositiveIntEnv("WORKER_CRAWL_CONCURRENCY", 1); - const archiveConcurrency = parsePositiveIntEnv("WORKER_ARCHIVE_CONCURRENCY", 1); - // Long crawl/upload/zip phases can starve lock renewal under heavy CPU pressure. - // Use conservative defaults to avoid duplicate retry attempts on healthy long-running jobs. - const lockDuration = parsePositiveIntEnv("WORKER_LOCK_DURATION_MS", 900000); - const stalledInterval = parsePositiveIntEnv("WORKER_STALLED_INTERVAL_MS", 120000); + const config = getWorkerRuntimeConfig(); const crawlWorker = new Worker("crawl-jobs", processCrawlJob, { connection: workerConnection, - concurrency: workerConcurrency, - lockDuration, - stalledInterval, + concurrency: config.crawlConcurrency, + lockDuration: config.lockDuration, + stalledInterval: config.stalledInterval, // Critical: Set to 0 to prevent BullMQ from auto-retrying stalled jobs // Stalled jobs are handled by orphan reconciliation - we want manual control maxStalledCount: 0, - // Disable automatic job recovery - we'll handle it via reconcileOrphanedCrawls - skipLockRenewal: true, + skipLockRenewal: config.skipLockRenewal, }); const archiveWorker = new Worker("archive-jobs", processArchiveJob, { connection: workerConnection, - concurrency: archiveConcurrency, - lockDuration, - stalledInterval, + concurrency: config.archiveConcurrency, + lockDuration: config.lockDuration, + stalledInterval: config.stalledInterval, maxStalledCount: 0, - skipLockRenewal: true, + skipLockRenewal: config.skipLockRenewal, }); - const reconcileIntervalMs = parsePositiveIntEnv("ORPHAN_CRAWL_RECONCILE_INTERVAL_MS", 120000); - void reconcileOrphanedCrawls().catch((error) => { + void reconcileOrphanedCrawls(config.orphanGraceMs).catch((error) => { console.error("[Worker] Failed to reconcile orphaned crawls:", error); }); const reconcileTimer = setInterval(() => { - void reconcileOrphanedCrawls().catch((error) => { + void reconcileOrphanedCrawls(config.orphanGraceMs).catch((error) => { console.error("[Worker] Failed to reconcile orphaned crawls:", error); }); - }, reconcileIntervalMs); + }, config.reconcileIntervalMs); reconcileTimer.unref(); attachWorkerLogging(crawlWorker, "crawl", reconcileTimer); attachWorkerLogging(archiveWorker, "archive", reconcileTimer); console.log( - `[Worker] Started crawl worker (concurrency=${workerConcurrency}) and archive worker (concurrency=${archiveConcurrency}, lockDuration=${lockDuration}ms)` + `[Worker] Started crawl worker (concurrency=${config.crawlConcurrency}) and archive worker (concurrency=${config.archiveConcurrency}, lockDuration=${config.lockDuration}ms)` ); return { diff --git a/services/worker/src/worker-config.test.ts b/services/worker/src/worker-config.test.ts new file mode 100644 index 0000000..be71944 --- /dev/null +++ b/services/worker/src/worker-config.test.ts @@ -0,0 +1,61 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { getWorkerRuntimeConfig } from "./worker-config.js"; + +describe("getWorkerRuntimeConfig", () => { + it("uses 10 minute stale-recovery defaults and keeps lock renewal enabled", () => { + const config = getWorkerRuntimeConfig({}); + + assert.equal(config.crawlConcurrency, 1); + assert.equal(config.archiveConcurrency, 1); + assert.equal(config.lockDuration, 10 * 60 * 1000); + assert.equal(config.stalledInterval, 120000); + assert.equal(config.orphanGraceMs, 10 * 60 * 1000); + assert.equal(config.reconcileIntervalMs, 120000); + assert.equal(config.skipLockRenewal, false); + }); + + it("respects positive environment overrides", () => { + const config = getWorkerRuntimeConfig({ + WORKER_CRAWL_CONCURRENCY: "2", + WORKER_ARCHIVE_CONCURRENCY: "3", + WORKER_LOCK_DURATION_MS: "720000", + WORKER_STALLED_INTERVAL_MS: "60000", + ORPHAN_CRAWL_GRACE_MS: "480000", + ORPHAN_CRAWL_RECONCILE_INTERVAL_MS: "30000", + WORKER_SKIP_LOCK_RENEWAL: "true", + } as NodeJS.ProcessEnv); + + assert.equal(config.crawlConcurrency, 2); + assert.equal(config.archiveConcurrency, 3); + assert.equal(config.lockDuration, 720000); + assert.equal(config.stalledInterval, 60000); + assert.equal(config.orphanGraceMs, 480000); + assert.equal(config.reconcileIntervalMs, 30000); + assert.equal(config.skipLockRenewal, true); + }); + + it("accepts common truthy values for skipLockRenewal", () => { + for (const raw of ["true", "TRUE", "1", "yes", "on"]) { + const config = getWorkerRuntimeConfig({ + WORKER_SKIP_LOCK_RENEWAL: raw, + } as NodeJS.ProcessEnv); + + assert.equal(config.skipLockRenewal, true, `expected ${raw} to enable skipLockRenewal`); + } + }); + + it("keeps skipLockRenewal disabled for falsey or absent values", () => { + for (const raw of [undefined, "false", "0", "no", "off", "unexpected"]) { + const config = getWorkerRuntimeConfig( + raw === undefined + ? {} + : ({ + WORKER_SKIP_LOCK_RENEWAL: raw, + } as NodeJS.ProcessEnv) + ); + + assert.equal(config.skipLockRenewal, false, `expected ${String(raw)} to keep skipLockRenewal disabled`); + } + }); +}); diff --git a/services/worker/src/worker-config.ts b/services/worker/src/worker-config.ts new file mode 100644 index 0000000..5eec6df --- /dev/null +++ b/services/worker/src/worker-config.ts @@ -0,0 +1,52 @@ +export type WorkerRuntimeConfig = { + crawlConcurrency: number; + archiveConcurrency: number; + lockDuration: number; + stalledInterval: number; + orphanGraceMs: number; + reconcileIntervalMs: number; + skipLockRenewal: boolean; +}; + +function readPositiveIntEnv(env: NodeJS.ProcessEnv, name: string, fallback: number): number { + const raw = env[name]; + if (!raw) { + return fallback; + } + + const parsed = Number.parseInt(raw, 10); + if (!Number.isFinite(parsed) || parsed <= 0) { + return fallback; + } + + return parsed; +} + +function readTruthyEnv(env: NodeJS.ProcessEnv, name: string): boolean { + const raw = env[name]; + if (!raw) { + return false; + } + + switch (raw.trim().toLowerCase()) { + case "true": + case "1": + case "yes": + case "on": + return true; + default: + return false; + } +} + +export function getWorkerRuntimeConfig(env: NodeJS.ProcessEnv = process.env): WorkerRuntimeConfig { + return { + crawlConcurrency: readPositiveIntEnv(env, "WORKER_CRAWL_CONCURRENCY", 1), + archiveConcurrency: readPositiveIntEnv(env, "WORKER_ARCHIVE_CONCURRENCY", 1), + lockDuration: readPositiveIntEnv(env, "WORKER_LOCK_DURATION_MS", 10 * 60 * 1000), + stalledInterval: readPositiveIntEnv(env, "WORKER_STALLED_INTERVAL_MS", 120000), + orphanGraceMs: readPositiveIntEnv(env, "ORPHAN_CRAWL_GRACE_MS", 10 * 60 * 1000), + reconcileIntervalMs: readPositiveIntEnv(env, "ORPHAN_CRAWL_RECONCILE_INTERVAL_MS", 120000), + skipLockRenewal: readTruthyEnv(env, "WORKER_SKIP_LOCK_RENEWAL"), + }; +}