fix: Abort dangling processing when crawler is aborted

[MohamedBassem]

An attempt for #1748

[coderabbitai]

Walkthrough

Adds abort-aware control flow and streamed, size-enforced asset downloads in crawlerWorker.ts; switches asset handling from in-memory buffers to temporary file pipelines, improves cleanup on abort/quota/error, and includes runNumber in generated jobId.

Changes

Cohort / File(s)	Summary of changes
Crawler abort + streaming downloads apps/workers/workers/crawlerWorker.ts	Added abortPromise helper and wired abort checks into navigation, wait-for-load, content extraction, screenshot capture, metadata extraction, and asset flows. Raced navigation and other critical async tasks against abort; added throwIfAborted checks.
Streamed asset pipeline & temp-file handling apps/workers/workers/crawlerWorker.ts	Introduced a Transform that tracks bytesRead, enforces a max asset size, validates content-length on flush, and pipes response body into a temp file. Replaced in-memory buffering with file-based asset saving (saveAssetFromFile) and updated quota checks to use streamed bytesRead. Ensured temp files are cleaned in finally blocks.
Archive / HTML storage changes apps/workers/workers/crawlerWorker.ts	archiveWebpage and HTML/content storage now write to temp files (via os.tmpdir) and perform cleanup when archiving is skipped, fails, or quota prevents storage. Preserved inline-storage alternatives for very large HTML.
Job and control-flow adjustments apps/workers/workers/crawlerWorker.ts	runCrawler jobId generation now includes runNumber. Enhanced content-type detection and propagated aborts through crawlAndParseUrl, including during content-type checks, image handling, downloads, and downstream processing.

Sequence Diagram(s)

sequenceDiagram
  autonumber
  participant W as Worker
  participant CW as CrawlerWorker
  participant P as BrowserPage
  participant N as HTTP Server
  participant FS as TempFileStore
  participant S as PersistentStore
  participant A as AbortSignal

  Note over CW,A #fff2cc: Abort-aware crawl with streamed asset downloads

  W->>CW: runCrawler(jobId + runNumber, A)
  CW->>P: page.goto(url)
  par Race navigation vs abort
    P-->>CW: navigate/loaded
    A-->>CW: abort signaled
  end
  alt Aborted
    CW-->>W: AbortError (cleanup temp files)
  else Loaded
    CW->>P: wait-for-load / extract HTML & metadata
    A-->>CW: abort?
    alt Aborted
      CW-->>W: AbortError
    else Continue
      CW->>P: optional screenshot (abort checked)
      CW->>N: fetch asset
      N-->>CW: response stream + headers
      CW->>FS: create temp file
      CW->>FS: pipeline(stream -> Transform[size-check] -> temp file)
      A-->>CW: abort?
      alt Aborted during download
        CW->>FS: cleanup temp file
        CW-->>W: AbortError
      else Download complete
        CW->>S: saveAssetFromFile(temp file, bytesRead)
        CW->>FS: cleanup temp file
        CW-->>W: result
      end
    end
  end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Pre-merge checks and finishing touches

❌ Failed checks (1 warning, 1 inconclusive)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	You can run @coderabbitai generate docstrings to improve docstring coverage.
Description Check	❓ Inconclusive	The description merely states “An attempt for #1748” and provides no information about what changes were made or why, making it too generic to convey the nature of the work in this pull request.	Please expand the description to include a brief summary of the key changes (for example, integration of an abortPromise helper, stream-based asset size enforcement, and cleanup of temporary files) so reviewers can quickly understand the scope and intent of the PR.

✅ Passed checks (1 passed)

Check name	Status	Explanation
Title Check	✅ Passed	The title succinctly and accurately describes the primary change—ensuring that any in-flight or “dangling” processing is aborted when the crawler’s abort signal fires—without extraneous details, making it clear to reviewers what the PR accomplishes.

✨ Finishing touches

• 📝 Generate Docstrings

🧪 Generate unit tests

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch abort-dangling-processing

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[cloudflare-workers-and-pages]

Deploying karakeep-landing with    Cloudflare Pages

Latest commit:	54b145e
Status:	✅  Deploy successful!
Preview URL:	https://1dc89b2b.karakeep-landing.pages.dev
Branch Preview URL:	https://abort-dangling-processing.karakeep-landing.pages.dev

View logs

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

apps/workers/workers/crawlerWorker.ts (1)

1006-1011: Abort guard keeps Promise.all pending forever

abortPromise never resolves; it only rejects when the signal fires. Including it in Promise.all([...]) means the array never settles on the non-aborted path, so crawlAndParseUrl hangs indefinitely and the worker never moves forward. Please race the combined work against the abort signal instead of awaiting a never-resolving promise.

-  const [meta, readableContent, screenshotAssetInfo] = await Promise.all([
-    extractMetadata(htmlContent, browserUrl, jobId),
-    extractReadableContent(htmlContent, browserUrl, jobId),
-    storeScreenshot(screenshot, userId, jobId),
-    abortPromise(abortSignal),
-  ]);
+  const abortableWork = Promise.all([
+    extractMetadata(htmlContent, browserUrl, jobId),
+    extractReadableContent(htmlContent, browserUrl, jobId),
+    storeScreenshot(screenshot, userId, jobId),
+  ]);
+  await Promise.race([abortableWork, abortPromise(abortSignal)]);
+  const [meta, readableContent, screenshotAssetInfo] = await abortableWork;

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 8dd84ef and 6f23df4.

📒 Files selected for processing (1)

• apps/workers/workers/crawlerWorker.ts (14 hunks)

🧰 Additional context used

📓 Path-based instructions (2)

**/*.{ts,tsx,js,jsx}

📄 CodeRabbit inference engine (AGENTS.md)

Use oxlint as the linter for JavaScript/TypeScript code

Files:

• apps/workers/workers/crawlerWorker.ts

apps/workers/**

📄 CodeRabbit inference engine (AGENTS.md)

Background workers code resides in apps/workers

Files:

• apps/workers/workers/crawlerWorker.ts

🧬 Code graph analysis (1)

apps/workers/workers/crawlerWorker.ts (4)

apps/workers/utils.ts (1)

• fetchWithProxy (62-71)

packages/shared/assetdb.ts (4)

• newAssetId (126-128)
• saveAssetFromFile (177-202)
• saveAssetFromFile (369-395)
• saveAssetFromFile (651-678)

packages/shared/tryCatch.ts (1)

• tryCatch (15-24)

packages/shared-server/src/services/quotaService.ts (1)

• QuotaService (22-93)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)

• GitHub Check: tests
• GitHub Check: typecheck

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

Compressed downloads always fail length validation

Response.body from undici is already decompressed. When a server sends a Content-Length for the compressed payload (e.g., gzip), bytesRead grows to the decompressed size, so the flush check throws Content length mismatch on every compressed asset. That breaks downloads for most real-world sites. Guard the comparison so it only runs when the header parses to a finite number and the response isn’t compressed.

-    const expectedLength = response.headers.get("content-length");
+    const rawContentLength = response.headers.get("content-length");
+    const parsedContentLength =
+      rawContentLength !== null ? Number(rawContentLength) : null;
+    const expectedLength =
+      parsedContentLength !== null &&
+      Number.isFinite(parsedContentLength) &&
+      !response.headers.get("content-encoding")
+        ? parsedContentLength
+        : null;
...
-        if (expectedLength && bytesRead !== Number(expectedLength)) {
+        if (expectedLength !== null && bytesRead !== expectedLength) {

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	const expectedLength = response.headers.get("content-length");
	const assetId = newAssetId();
	assetPath = path.join(os.tmpdir(), assetId);
	let bytesRead = 0;
	const contentLengthEnforcer = new Transform({
	transform(chunk, _, callback) {
	bytesRead += chunk.length;
	if (abortSignal.aborted) {
	callback(new Error("AbortError"));
	} else if (bytesRead > serverConfig.maxAssetSizeMb * 1024 * 1024) {
	callback(
	new Error(
	`Content length exceeds maximum allowed size: ${serverConfig.maxAssetSizeMb}MB`,
	),
	);
	} else {
	callback(null, chunk); // pass data along unchanged
	}
	},
	flush(callback) {
	if (expectedLength && bytesRead !== Number(expectedLength)) {
	callback(
	new Error(
	`Content length mismatch: expected ${expectedLength}, got ${bytesRead}`,
	),
	);
	} else {
	callback();
	}
	},
	});
	const rawContentLength = response.headers.get("content-length");
	const parsedContentLength =
	rawContentLength !== null ? Number(rawContentLength) : null;
	const expectedLength =
	parsedContentLength !== null &&
	Number.isFinite(parsedContentLength) &&
	!response.headers.get("content-encoding")
	? parsedContentLength
	: null;
	const assetId = newAssetId();
	assetPath = path.join(os.tmpdir(), assetId);
	let bytesRead = 0;
	const contentLengthEnforcer = new Transform({
	transform(chunk, _, callback) {
	bytesRead += chunk.length;
	if (abortSignal.aborted) {
	callback(new Error("AbortError"));
	} else if (bytesRead > serverConfig.maxAssetSizeMb * 1024 * 1024) {
	callback(
	new Error(
	`Content length exceeds maximum allowed size: ${serverConfig.maxAssetSizeMb}MB`,
	),
	);
	} else {
	callback(null, chunk); // pass data along unchanged
	}
	},
	flush(callback) {
	if (expectedLength !== null && bytesRead !== expectedLength) {
	callback(
	new Error(
	`Content length mismatch: expected ${expectedLength}, got ${bytesRead}`,
	),
	);
	} else {
	callback();
	}
	},
	});

[coderabbitai]

Actionable comments posted: 1

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 6f23df4 and 2e16ff4.

📒 Files selected for processing (1)

• apps/workers/workers/crawlerWorker.ts (15 hunks)

🧰 Additional context used

📓 Path-based instructions (2)

**/*.{ts,tsx,js,jsx}

📄 CodeRabbit inference engine (AGENTS.md)

Use oxlint as the linter for JavaScript/TypeScript code

Files:

• apps/workers/workers/crawlerWorker.ts

apps/workers/**

📄 CodeRabbit inference engine (AGENTS.md)

Background workers code resides in apps/workers

Files:

• apps/workers/workers/crawlerWorker.ts

🧬 Code graph analysis (1)

apps/workers/workers/crawlerWorker.ts (4)

apps/workers/utils.ts (1)

• fetchWithProxy (62-71)

packages/shared/assetdb.ts (4)

• newAssetId (126-128)
• saveAssetFromFile (177-202)
• saveAssetFromFile (369-395)
• saveAssetFromFile (651-678)

packages/shared/tryCatch.ts (1)

• tryCatch (15-24)

packages/shared-server/src/services/quotaService.ts (1)

• QuotaService (22-93)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)

• GitHub Check: tests
• GitHub Check: typecheck

[coderabbitai]

Actionable comments posted: 2

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 2e16ff4 and 8bebdcc.

📒 Files selected for processing (1)

• apps/workers/workers/crawlerWorker.ts (15 hunks)

🧰 Additional context used

📓 Path-based instructions (2)

**/*.{ts,tsx,js,jsx}

📄 CodeRabbit inference engine (AGENTS.md)

Use oxlint as the linter for JavaScript/TypeScript code

Files:

• apps/workers/workers/crawlerWorker.ts

apps/workers/**

📄 CodeRabbit inference engine (AGENTS.md)

Background workers code resides in apps/workers

Files:

• apps/workers/workers/crawlerWorker.ts

🧬 Code graph analysis (1)

apps/workers/workers/crawlerWorker.ts (4)

apps/workers/utils.ts (1)

• fetchWithProxy (62-71)

packages/shared/assetdb.ts (4)

• newAssetId (126-128)
• saveAssetFromFile (177-202)
• saveAssetFromFile (369-395)
• saveAssetFromFile (651-678)

packages/shared/tryCatch.ts (1)

• tryCatch (15-24)

packages/shared-server/src/services/quotaService.ts (1)

• QuotaService (22-93)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)

• GitHub Check: tests
• GitHub Check: typecheck

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

Fix: Convert fetch body to a Node stream before piping

Line 683: stream.pipeline expects a Node Readable, but fetchWithProxy returns an Undici/WHATWG ReadableStream. Passing it directly throws ERR_INVALID_ARG_TYPE, so every download aborts at runtime. Please wrap the body with Readable.fromWeb(...) (and import Readable) before piping it into the temp file.

-import { pipeline, Transform } from "stream";
+import { pipeline, Transform, Readable } from "stream";
…
-    await streamPipeline(
-      response.body,
+    await streamPipeline(
+      Readable.fromWeb(response.body),
       contentLengthEnforcer,
       fsSync.createWriteStream(assetPath),
     );

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	await streamPipeline(
	response.body,
	contentLengthEnforcer,
	fsSync.createWriteStream(assetPath),
	);
	// At the top of the file, include Readable
	import { pipeline, Transform, Readable } from "stream";
	…
	// In the download pipeline, convert the web ReadableStream to a Node Readable
	- await streamPipeline(
	await streamPipeline(
	Readable.fromWeb(response.body),
	contentLengthEnforcer,
	fsSync.createWriteStream(assetPath),
	);

🤖 Prompt for AI Agents

In apps/workers/workers/crawlerWorker.ts around lines 681 to 685, the code
passes the WHATWG/Undici Response.body directly into stream.pipeline which
expects a Node Readable, causing ERR_INVALID_ARG_TYPE; wrap the fetch body with
Readable.fromWeb(response.body) before piping and add the Node stream import
(import { Readable } from 'stream' or equivalent CommonJS require) at the top of
the file so streamPipeline receives a proper Node Readable.

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

Propagate aborts instead of treating them as download failures

Line 715: When the abort signal fires, either fetch or the transform surfaces an AbortError, but this catch block logs it and returns null. In the PDF/image path (handleAsAssetBookmark), that means the crawler exits normally instead of honoring the abort—which is the regression this PR is trying to avoid. Please rethrow abort-driven failures (e.g., check abortSignal.aborted or e?.name === "AbortError" and throw the abort reason) so callers can stop the job.

   } catch (e) {
-    logger.error(
-      `[Crawler][${jobId}] Failed to download and store ${fileType}: ${e}`,
-    );
-    return null;
+    if (
+      abortSignal.aborted ||
+      (e instanceof Error && e.name === "AbortError")
+    ) {
+      throw abortSignal.reason ?? e;
+    }
+    logger.error(
+      `[Crawler][${jobId}] Failed to download and store ${fileType}: ${e}`,
+    );
+    return null;
   } finally {

🤖 Prompt for AI Agents

In apps/workers/workers/crawlerWorker.ts around lines 713-717, the catch
currently logs all errors from fetch/transform and returns null which swallows
abort-driven failures; change the handler to detect aborts (e.g., if
(abortSignal?.aborted) or if (e?.name === "AbortError")) and rethrow the abort
(preferably throw abortSignal.reason || e) so the caller can stop the job, while
still logging non-abort errors and returning null for those.

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

apps/workers/workers/crawlerWorker.ts (1)

746-747: Verify temporary file path generation

Using os.tmpdir() with just the assetId as filename could potentially cause conflicts if the same ID is generated multiple times.

Consider using a more unique temporary filename:

-  const assetPath = path.join(os.tmpdir(), assetId);
+  const assetPath = path.join(os.tmpdir(), `karakeep_archive_${assetId}.html`);

This adds a prefix and extension for better file management and debugging.

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 8bebdcc and 54b145e.

📒 Files selected for processing (1)

• apps/workers/workers/crawlerWorker.ts (15 hunks)

🧰 Additional context used

📓 Path-based instructions (2)

**/*.{ts,tsx,js,jsx}

📄 CodeRabbit inference engine (AGENTS.md)

Use oxlint as the linter for JavaScript/TypeScript code

Files:

• apps/workers/workers/crawlerWorker.ts

apps/workers/**

📄 CodeRabbit inference engine (AGENTS.md)

Background workers code resides in apps/workers

Files:

• apps/workers/workers/crawlerWorker.ts

🧬 Code graph analysis (1)

apps/workers/workers/crawlerWorker.ts (4)

apps/workers/utils.ts (1)

• fetchWithProxy (62-71)

packages/shared/assetdb.ts (4)

• newAssetId (126-128)
• saveAssetFromFile (177-202)
• saveAssetFromFile (369-395)
• saveAssetFromFile (651-678)

packages/shared/tryCatch.ts (1)

• tryCatch (15-24)

packages/shared-server/src/services/quotaService.ts (1)

• QuotaService (22-93)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)

• GitHub Check: tests
• GitHub Check: typecheck

🔇 Additional comments (13)

apps/workers/workers/crawlerWorker.ts (13)

457-463: Verify null handling for navigation response

When abortPromise wins the race on line 462, response becomes null, but lines 523-524 still access response?.status() which correctly handles null. This is good defensive programming.

The null-safe access pattern response?.status() ?? 0 properly handles the abort case where response is null. Well done!

---

476-477: Good practice: Immediate abort check after async operations

Adding abortSignal.throwIfAborted() after the wait ensures the crawler stops immediately when aborted, preventing unnecessary processing.

---

715-719: Good practice: Cleanup temporary files in finally block

The finally block ensures temporary files are cleaned up even on errors, preventing disk space leaks.

---

1007-1018: Effective abort handling with race conditions

The implementation properly races extractors against abort and then awaits the work to get results or throw on abort. The pattern ensures work completes or aborts cleanly.

Excellent pattern for handling concurrent operations with abort support! The code races the work against abort, then awaits the work promise to either get results or propagate any errors that occurred during execution.

---

1174-1174: Good improvement: Include runNumber in jobId for uniqueness

Adding runNumber to the jobId ensures unique identifiers across retries, improving log traceability.

---

1201-1201: Consistent abort checking throughout the flow

The abort check after getContentType ensures the job stops if aborted during content-type detection.

Good placement of abort check after the async operation. Based on learnings

---

783-783: Good error handling: Cleanup on quota failure

Properly cleaning up the temporary file when quota check fails prevents orphaned files.

---

635-636: Stream-based architecture improves memory efficiency

The transition from in-memory buffers to stream-based file operations with temporary files is a significant improvement for handling large assets without memory exhaustion.

Excellent architectural improvement! The stream-based approach with temporary files prevents memory issues when downloading large assets.

Also applies to: 654-656

---

641-643: Fix needed: Convert WHATWG stream to Node stream

The fetchWithProxy returns a WHATWG ReadableStream, but pipeline expects a Node.js Readable stream. This will cause a runtime error.

---

657-677: Verify content-length validation for compressed responses

The content-length enforcer doesn't account for compressed responses where the Content-Length header refers to the compressed size, but bytesRead tracks decompressed bytes.

---

710-719: Critical: Propagate abort errors instead of swallowing them

The catch block logs all errors and returns null, which prevents abort propagation. This defeats the purpose of abort handling.

---

679-683: Stream conversion required before pipeline

The WHATWG stream needs to be converted to a Node.js stream.

Import Readable from stream and convert the response body:

-import { Transform } from "stream";
+import { Transform, Readable } from "stream";
...
    await pipeline(
-      response.body,
+      Readable.fromWeb(response.body),
      contentLengthEnforcer,
      fsSync.createWriteStream(assetPath),
    );

---

78-101: Critical fix needed: Prevent unhandled rejections in abortPromise

The current implementation has a critical issue where unhandled rejections can occur when the promise is rejected after being raced but before being awaited. The rejected promise created on line 80 isn't properly caught, and the main promise (line 87) can still reject after its race resolves.

Apply this fix to properly handle rejections and clean up listeners:

function abortPromise(signal: AbortSignal): Promise<never> {
  if (signal.aborted) {
-    const p = Promise.reject(signal.reason ?? new Error("AbortError"));
-    p.catch(() => {
-      /* empty */
-    }); // suppress unhandledRejection if not awaited
-    return p;
+    return Promise.reject(signal.reason ?? new Error("AbortError"));
  }

-  const p = new Promise<never>((_, reject) => {
-    signal.addEventListener(
-      "abort",
-      () => {
-        reject(signal.reason ?? new Error("AbortError"));
-      },
-      { once: true },
-    );
-  });
+  let onAbort: (() => void) | undefined;
+  const promise = new Promise<never>((_, reject) => {
+    onAbort = () => {
+      signal.removeEventListener("abort", onAbort!);
+      reject(signal.reason ?? new Error("AbortError"));
+    };
+    signal.addEventListener("abort", onAbort, { once: true });
+  });

-  p.catch(() => {
-    /* empty */
-  });
-  return p;
+  // Attach a no-op catch handler to prevent unhandled rejection warnings
+  // when the promise is raced and loses
+  promise.catch(() => {});
+  return promise;
}

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Potential issue: Empty buffer on abort may cause issues downstream

When the abort promise wins on line 506, it returns an empty buffer Buffer.from(""). This could be misleading as it appears successful but with empty data.

Consider throwing the abort error instead of returning an empty buffer:

-          abortPromise(abortSignal).then(() => Buffer.from("")),
+          abortPromise(abortSignal),

This way, the error handling on line 510 will properly detect the abort.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	abortPromise(abortSignal).then(() => Buffer.from("")),
	]),
	);
	abortSignal.throwIfAborted();
	abortPromise(abortSignal),
	]),
	);
	abortSignal.throwIfAborted();

🤖 Prompt for AI Agents

In apps/workers/workers/crawlerWorker.ts around lines 506 to 509, returning an
empty Buffer when the abort promise wins hides the abort as a successful empty
result; change the abort branch so it throws/rejects the abort error instead of
resolving Buffer.from("") (e.g. have abortPromise reject with the AbortError or
propagate abortSignal.reason), so the race will surface the abort as an
exception and existing abortSignal.throwIfAborted() / surrounding error handling
will detect it.