fix: harden crawl filename handling and stale recovery

.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

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",

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");
+  });
 });

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()

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<CrawlJobData>) {
   });
 }
 
-async function reconcileOrphanedCrawls(): Promise<void> {
-  const orphanGraceMs = parsePositiveIntEnv("ORPHAN_CRAWL_GRACE_MS", 5 * 60 * 1000);
+async function reconcileOrphanedCrawls(orphanGraceMs = getWorkerRuntimeConfig().orphanGraceMs): Promise<void> {
   const cutoff = new Date(Date.now() - orphanGraceMs);
 
   const possiblyOrphaned = await db.query.crawls.findMany({
@@ -1137,51 +1137,44 @@ function attachWorkerLogging<T>(worker: Worker<T>, 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<CrawlJobData>("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<ArchiveJobData>("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 {

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`);
+    }
+  });
+});

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"),
+  };
+}