Audit crawl: conservative OOM retries + window carry-over across chunks (#517)
* Audit crawl: retry chunks conservatively and stop re-learning the window per chunk Production audits on heavy-page sites (500-700KB/page) died with exceededMemory in the first crawl chunk, and the single step retry re-ran the exact same profile and died again. - claimChunk now reports isRetry (leftover leases from a dead attempt); a retried chunk crawls under RETRY_CRAWL_WINDOW (start 3, max 5, halved byte budget) instead of restarting at window 10. - The adapted window carries across chunks via durable step results, so every ~200 pages no longer re-spikes to the initial window. - First persist sub-batch shrinks to 5 pages so the byte bound sees the site's page weight before a full 25-page batch is in flight. - In-flight HTML budget halved to 8 MiB (16 MiB never constrained the observed ~650KB pages: bound was 25, above the 20 max); growth now requires a full 25-page sample. - Parse-time caps: 1,000 extracted links/images per page so mega-menu and crawler-trap pages can't bloat retained persist batches. * Guard endWindow for instances replaying pre-deploy step results * ci: un-export internal-only interfaces (knip)
This commit is contained in:
parent
4cd321e269
commit
e5e961bf48
@ -26,6 +26,17 @@ export interface ClaimedUrl {
|
|||||||
inSitemap: boolean;
|
inSitemap: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
interface ClaimedChunk {
|
||||||
|
urls: ClaimedUrl[];
|
||||||
|
/**
|
||||||
|
* True when a prior attempt of this chunk already claimed work — i.e. the
|
||||||
|
* caller is a workflow step retry after that attempt died mid-crawl. The
|
||||||
|
* crawler uses this to retry far more conservatively (the usual cause of a
|
||||||
|
* dead attempt is exceededMemory).
|
||||||
|
*/
|
||||||
|
isRetry: boolean;
|
||||||
|
}
|
||||||
|
|
||||||
export interface FrontierStats {
|
export interface FrontierStats {
|
||||||
/** Pages attempted (crawled or errored) so far. */
|
/** Pages attempted (crawled or errored) so far. */
|
||||||
attempted: number;
|
attempted: number;
|
||||||
@ -147,12 +158,12 @@ export class AuditScratchpad extends DurableObject {
|
|||||||
* had leased (link-discovered URLs drain before sitemap-only ones, FIFO
|
* had leased (link-discovered URLs drain before sitemap-only ones, FIFO
|
||||||
* within each class — same ordering as the old in-memory queues).
|
* within each class — same ordering as the old in-memory queues).
|
||||||
*/
|
*/
|
||||||
async claimChunk(chunkNo: number, limit: number): Promise<ClaimedUrl[]> {
|
async claimChunk(chunkNo: number, limit: number): Promise<ClaimedChunk> {
|
||||||
const existing = this.selectClaimed(
|
const existing = this.selectClaimed(
|
||||||
`SELECT url, depth, in_sitemap FROM frontier WHERE state = 'leased' AND chunk_no = ?`,
|
`SELECT url, depth, in_sitemap FROM frontier WHERE state = 'leased' AND chunk_no = ?`,
|
||||||
chunkNo,
|
chunkNo,
|
||||||
);
|
);
|
||||||
if (existing.length > 0) return existing;
|
if (existing.length > 0) return { urls: existing, isRetry: true };
|
||||||
// A retried step whose earlier attempt already crawled this chunk's
|
// A retried step whose earlier attempt already crawled this chunk's
|
||||||
// leases must not claim a fresh set under the same chunk number — that
|
// leases must not claim a fresh set under the same chunk number — that
|
||||||
// would duplicate work and overshoot the page budget.
|
// would duplicate work and overshoot the page budget.
|
||||||
@ -164,8 +175,8 @@ export class AuditScratchpad extends DurableObject {
|
|||||||
chunkNo,
|
chunkNo,
|
||||||
)
|
)
|
||||||
.one();
|
.one();
|
||||||
if (done.n > 0) return [];
|
if (done.n > 0) return { urls: [], isRetry: true };
|
||||||
if (limit <= 0) return [];
|
if (limit <= 0) return { urls: [], isRetry: false };
|
||||||
|
|
||||||
const fresh = this.selectClaimed(
|
const fresh = this.selectClaimed(
|
||||||
`SELECT url, depth, in_sitemap FROM frontier WHERE state = 'pending'
|
`SELECT url, depth, in_sitemap FROM frontier WHERE state = 'pending'
|
||||||
@ -179,7 +190,7 @@ export class AuditScratchpad extends DurableObject {
|
|||||||
row.url,
|
row.url,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
return fresh;
|
return { urls: fresh, isRetry: false };
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Persist one crawled sub-batch: completions, mirror rows, links, frontier. */
|
/** Persist one crawled sub-batch: completions, mirror rows, links, frontier. */
|
||||||
|
|||||||
@ -1,5 +1,8 @@
|
|||||||
import { describe, expect, it } from "vitest";
|
import { describe, expect, it } from "vitest";
|
||||||
import { adjustCrawlWindow } from "@/server/lib/audit/crawl-window";
|
import {
|
||||||
|
adjustCrawlWindow,
|
||||||
|
RETRY_CRAWL_WINDOW,
|
||||||
|
} from "@/server/lib/audit/crawl-window";
|
||||||
import type {
|
import type {
|
||||||
CrawledPageResult,
|
CrawledPageResult,
|
||||||
PageFetchClass,
|
PageFetchClass,
|
||||||
@ -79,11 +82,30 @@ describe("adjustCrawlWindow", () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
it("caps the window so heavy pages stay inside the byte budget", () => {
|
it("caps the window so heavy pages stay inside the byte budget", () => {
|
||||||
// 1 MiB average pages: 16 MiB budget / 1 MiB = window of 16.
|
// 1 MiB average pages: 8 MiB budget / 1 MiB = window of 8.
|
||||||
const recent = Array.from({ length: 25 }, () =>
|
const recent = Array.from({ length: 25 }, () =>
|
||||||
page("ok", 300, 1024 * 1024),
|
page("ok", 300, 1024 * 1024),
|
||||||
);
|
);
|
||||||
expect(adjustCrawlWindow(20, recent)).toBe(16);
|
expect(adjustCrawlWindow(20, recent)).toBe(8);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not grow on a small sample, but still applies the byte bound", () => {
|
||||||
|
const fastSmall = Array.from({ length: 5 }, () => page("ok", 400));
|
||||||
|
expect(adjustCrawlWindow(10, fastSmall)).toBe(10);
|
||||||
|
const fastHeavy = Array.from({ length: 5 }, () =>
|
||||||
|
page("ok", 400, 1024 * 1024),
|
||||||
|
);
|
||||||
|
expect(adjustCrawlWindow(10, fastHeavy)).toBe(8);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("retry limits keep the window small even on a clean, fast site", () => {
|
||||||
|
const recent = Array.from({ length: 25 }, () => page("ok", 400));
|
||||||
|
expect(adjustCrawlWindow(3, recent, RETRY_CRAWL_WINDOW)).toBe(5);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("retry limits shrink below the normal minimum on trouble", () => {
|
||||||
|
const recent = Array.from({ length: 10 }, () => page("error", 15_000));
|
||||||
|
expect(adjustCrawlWindow(3, recent, RETRY_CRAWL_WINDOW)).toBe(2);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("keeps the byte bound at the minimum window even for huge pages", () => {
|
it("keeps the byte bound at the minimum window even for huge pages", () => {
|
||||||
|
|||||||
@ -7,12 +7,11 @@ import type { CrawledPageResult } from "@/server/lib/audit/types";
|
|||||||
* slowly (politeness toward struggling or defensive sites) and grows when
|
* slowly (politeness toward struggling or defensive sites) and grows when
|
||||||
* the site answers fast.
|
* the site answers fast.
|
||||||
*/
|
*/
|
||||||
export const INITIAL_CRAWL_WINDOW = 10;
|
interface CrawlWindowLimits {
|
||||||
const MIN_WINDOW = 5;
|
/** Window size a chunk starts with, before any observations. */
|
||||||
const MAX_WINDOW = 20;
|
initial: number;
|
||||||
const SLOW_RESPONSE_MS = 10_000;
|
min: number;
|
||||||
const FAST_RESPONSE_MS = 1_500;
|
max: number;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Total HTML the in-flight window may buffer at once. Each in-flight page
|
* Total HTML the in-flight window may buffer at once. Each in-flight page
|
||||||
* holds its body (decoded to a UTF-16 string, roughly doubling it), and the
|
* holds its body (decoded to a UTF-16 string, roughly doubling it), and the
|
||||||
@ -20,19 +19,57 @@ const FAST_RESPONSE_MS = 1_500;
|
|||||||
* worker — production audits died with exceededMemory when a fast site let
|
* worker — production audits died with exceededMemory when a fast site let
|
||||||
* the window grow unchecked.
|
* the window grow unchecked.
|
||||||
*/
|
*/
|
||||||
const IN_FLIGHT_HTML_BUDGET_BYTES = 16 * 1024 * 1024;
|
budgetBytes: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
export const CRAWL_WINDOW: CrawlWindowLimits = {
|
||||||
|
initial: 10,
|
||||||
|
min: 5,
|
||||||
|
max: 20,
|
||||||
|
budgetBytes: 8 * 1024 * 1024,
|
||||||
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Limits for a chunk whose earlier attempt died mid-crawl — in production
|
||||||
|
* almost always exceededMemory on a heavy-page site. The retry is the
|
||||||
|
* chunk's last attempt (retry limit 1), so it must not re-run the exact
|
||||||
|
* memory profile that just killed the isolate.
|
||||||
|
*/
|
||||||
|
export const RETRY_CRAWL_WINDOW: CrawlWindowLimits = {
|
||||||
|
initial: 3,
|
||||||
|
min: 2,
|
||||||
|
max: 5,
|
||||||
|
budgetBytes: 4 * 1024 * 1024,
|
||||||
|
};
|
||||||
|
|
||||||
|
const SLOW_RESPONSE_MS = 10_000;
|
||||||
|
const FAST_RESPONSE_MS = 1_500;
|
||||||
/** Floor for the observed page size so tiny-page sites can't void the bound. */
|
/** Floor for the observed page size so tiny-page sites can't void the bound. */
|
||||||
const MIN_ASSUMED_PAGE_BYTES = 64 * 1024;
|
const MIN_ASSUMED_PAGE_BYTES = 64 * 1024;
|
||||||
|
/**
|
||||||
|
* Growth requires a full-size sample. The first persist sub-batch is small
|
||||||
|
* (so the byte bound reacts to heavy pages early), and a handful of fast
|
||||||
|
* pages proves too little to widen the window.
|
||||||
|
*/
|
||||||
|
const GROWTH_MIN_SAMPLE = 25;
|
||||||
|
|
||||||
|
export function clampCrawlWindow(
|
||||||
|
size: number,
|
||||||
|
limits: CrawlWindowLimits,
|
||||||
|
): number {
|
||||||
|
return Math.min(Math.max(size, limits.min), limits.max);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Adapt the window to the last persisted sub-batch. Shrinks on trouble
|
* Adapt the window to the last persisted sub-batch. Shrinks on trouble
|
||||||
* (errors, blocks, very slow responses), grows only on a clean and mostly
|
* (errors, blocks, very slow responses), grows only on a clean, mostly fast,
|
||||||
* fast batch, and is always capped so the batch's average page size times
|
* full-size batch, and is always capped so the batch's average page size
|
||||||
* the window stays inside the in-flight byte budget.
|
* times the window stays inside the in-flight byte budget.
|
||||||
*/
|
*/
|
||||||
export function adjustCrawlWindow(
|
export function adjustCrawlWindow(
|
||||||
windowSize: number,
|
windowSize: number,
|
||||||
recent: CrawledPageResult[],
|
recent: CrawledPageResult[],
|
||||||
|
limits: CrawlWindowLimits = CRAWL_WINDOW,
|
||||||
): number {
|
): number {
|
||||||
if (recent.length === 0) return windowSize;
|
if (recent.length === 0) return windowSize;
|
||||||
const troubled = recent.filter(
|
const troubled = recent.filter(
|
||||||
@ -42,15 +79,19 @@ export function adjustCrawlWindow(
|
|||||||
).length;
|
).length;
|
||||||
let next = windowSize;
|
let next = windowSize;
|
||||||
if (troubled * 3 >= recent.length) {
|
if (troubled * 3 >= recent.length) {
|
||||||
next = Math.max(MIN_WINDOW, Math.floor(windowSize / 2));
|
next = Math.max(limits.min, Math.floor(windowSize / 2));
|
||||||
} else {
|
} else {
|
||||||
const fast = recent.filter(
|
const fast = recent.filter(
|
||||||
(page) =>
|
(page) =>
|
||||||
page.fetchClass === "ok" &&
|
page.fetchClass === "ok" &&
|
||||||
(page.responseTimeMs ?? Infinity) <= FAST_RESPONSE_MS,
|
(page.responseTimeMs ?? Infinity) <= FAST_RESPONSE_MS,
|
||||||
).length;
|
).length;
|
||||||
if (troubled === 0 && fast * 2 >= recent.length) {
|
if (
|
||||||
next = Math.min(MAX_WINDOW, windowSize + 5);
|
troubled === 0 &&
|
||||||
|
fast * 2 >= recent.length &&
|
||||||
|
recent.length >= GROWTH_MIN_SAMPLE
|
||||||
|
) {
|
||||||
|
next = Math.min(limits.max, windowSize + 5);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -59,8 +100,8 @@ export function adjustCrawlWindow(
|
|||||||
MIN_ASSUMED_PAGE_BYTES,
|
MIN_ASSUMED_PAGE_BYTES,
|
||||||
);
|
);
|
||||||
const byteBound = Math.max(
|
const byteBound = Math.max(
|
||||||
MIN_WINDOW,
|
limits.min,
|
||||||
Math.floor(IN_FLIGHT_HTML_BUDGET_BYTES / avgPageBytes),
|
Math.floor(limits.budgetBytes / avgPageBytes),
|
||||||
);
|
);
|
||||||
return Math.min(next, byteBound);
|
return Math.min(next, byteBound);
|
||||||
}
|
}
|
||||||
|
|||||||
@ -206,3 +206,24 @@ describe("analyzeHtml parity with the DOM reference", () => {
|
|||||||
</body>`);
|
</body>`);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("analyzeHtml extraction caps", () => {
|
||||||
|
it("caps links and images per page", () => {
|
||||||
|
const links = Array.from(
|
||||||
|
{ length: 1_100 },
|
||||||
|
(_, i) => `<a href="/p/${i}">link ${i}</a>`,
|
||||||
|
).join("");
|
||||||
|
const images = Array.from(
|
||||||
|
{ length: 1_100 },
|
||||||
|
(_, i) => `<img src="/i/${i}.png">`,
|
||||||
|
).join("");
|
||||||
|
const analysis = analyzeHtml(
|
||||||
|
`<body>${links}${images}</body>`,
|
||||||
|
"https://example.com/",
|
||||||
|
200,
|
||||||
|
100,
|
||||||
|
);
|
||||||
|
expect(analysis.links).toHaveLength(1_000);
|
||||||
|
expect(analysis.images).toHaveLength(1_000);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
@ -26,6 +26,15 @@ const HEADING_LEVELS: Record<string, number> = {
|
|||||||
h6: 6,
|
h6: 6,
|
||||||
};
|
};
|
||||||
const MAX_ANCHOR_CHARS = 200;
|
const MAX_ANCHOR_CHARS = 200;
|
||||||
|
/**
|
||||||
|
* Per-page caps on the extracted collections. Crawler-trap and mega-menu
|
||||||
|
* pages can carry thousands of links/images per page, and crawled pages sit
|
||||||
|
* in memory in 25-page persist batches — uncapped collections were part of
|
||||||
|
* the audit engine's exceededMemory profile. Counts derived from these
|
||||||
|
* arrays saturate at the cap on such pathological pages.
|
||||||
|
*/
|
||||||
|
const MAX_EXTRACTED_LINKS = 1_000;
|
||||||
|
const MAX_EXTRACTED_IMAGES = 1_000;
|
||||||
|
|
||||||
interface OpenAnchor {
|
interface OpenAnchor {
|
||||||
href: string;
|
href: string;
|
||||||
@ -103,6 +112,7 @@ export function analyzeHtml(
|
|||||||
if (!openAnchor) return;
|
if (!openAnchor) return;
|
||||||
const { href, rel, text } = openAnchor;
|
const { href, rel, text } = openAnchor;
|
||||||
openAnchor = null;
|
openAnchor = null;
|
||||||
|
if (linksByTarget.size >= MAX_EXTRACTED_LINKS) return;
|
||||||
const resolved = normalizeUrl(href, pageUrl);
|
const resolved = normalizeUrl(href, pageUrl);
|
||||||
if (!resolved || linksByTarget.has(resolved)) return;
|
if (!resolved || linksByTarget.has(resolved)) return;
|
||||||
const anchor = text
|
const anchor = text
|
||||||
@ -148,10 +158,12 @@ export function analyzeHtml(
|
|||||||
handleLinkTag(attribs);
|
handleLinkTag(attribs);
|
||||||
break;
|
break;
|
||||||
case "img":
|
case "img":
|
||||||
|
if (images.length < MAX_EXTRACTED_IMAGES) {
|
||||||
images.push({
|
images.push({
|
||||||
src: attribs["src"] ?? null,
|
src: attribs["src"] ?? null,
|
||||||
alt: "alt" in attribs ? attribs["alt"] : null,
|
alt: "alt" in attribs ? attribs["alt"] : null,
|
||||||
});
|
});
|
||||||
|
}
|
||||||
break;
|
break;
|
||||||
case "script":
|
case "script":
|
||||||
if (attribs["type"] === "application/ld+json") {
|
if (attribs["type"] === "application/ld+json") {
|
||||||
|
|||||||
@ -15,7 +15,9 @@ import {
|
|||||||
import { AuditProgressKV } from "@/server/lib/audit/progress-kv";
|
import { AuditProgressKV } from "@/server/lib/audit/progress-kv";
|
||||||
import {
|
import {
|
||||||
adjustCrawlWindow,
|
adjustCrawlWindow,
|
||||||
INITIAL_CRAWL_WINDOW,
|
clampCrawlWindow,
|
||||||
|
CRAWL_WINDOW,
|
||||||
|
RETRY_CRAWL_WINDOW,
|
||||||
} from "@/server/lib/audit/crawl-window";
|
} from "@/server/lib/audit/crawl-window";
|
||||||
import { crawlPage } from "@/server/workflows/site-audit-workflow-helpers";
|
import { crawlPage } from "@/server/workflows/site-audit-workflow-helpers";
|
||||||
import { pgStep } from "@/server/workflows/pgStep";
|
import { pgStep } from "@/server/workflows/pgStep";
|
||||||
@ -34,6 +36,14 @@ const CHUNK_TARGET_PAGES = 200;
|
|||||||
const CHUNK_SOFT_DEADLINE_MS = 90_000;
|
const CHUNK_SOFT_DEADLINE_MS = 90_000;
|
||||||
/** Crawled pages are persisted in sub-batches of this size. */
|
/** Crawled pages are persisted in sub-batches of this size. */
|
||||||
const PERSIST_BATCH_SIZE = 25;
|
const PERSIST_BATCH_SIZE = 25;
|
||||||
|
/**
|
||||||
|
* The first sub-batch of a chunk is deliberately small: the crawl window
|
||||||
|
* only adapts when a sub-batch persists, and on a heavy-page site a full
|
||||||
|
* 25-page batch crawled at the starting window was already enough to
|
||||||
|
* exceed the isolate's memory. Five pages tell the byte bound what the
|
||||||
|
* site's pages weigh before the window commits to more.
|
||||||
|
*/
|
||||||
|
const FIRST_PERSIST_BATCH_SIZE = 5;
|
||||||
/**
|
/**
|
||||||
* Stop launching new fetches while more than this many persist sub-batches
|
* Stop launching new fetches while more than this many persist sub-batches
|
||||||
* are waiting: persistence is sequential, so when the DB falls behind a fast
|
* are waiting: persistence is sequential, so when the DB falls behind a fast
|
||||||
@ -91,6 +101,11 @@ export async function runCrawlPhase(
|
|||||||
let attemptedTotal = 0;
|
let attemptedTotal = 0;
|
||||||
let pending = params.seededCount;
|
let pending = params.seededCount;
|
||||||
let zeroProgressChunks = 0;
|
let zeroProgressChunks = 0;
|
||||||
|
// The adapted window carries across chunks via durable step results:
|
||||||
|
// without this every chunk restarted at the initial window and re-learned
|
||||||
|
// the site's page weight the hard way — on heavy-page sites that meant an
|
||||||
|
// exceededMemory death every ~200 pages.
|
||||||
|
let windowHint = CRAWL_WINDOW.initial;
|
||||||
|
|
||||||
while (pending > 0 && attemptedTotal < params.maxPages) {
|
while (pending > 0 && attemptedTotal < params.maxPages) {
|
||||||
chunkNo += 1;
|
chunkNo += 1;
|
||||||
@ -103,6 +118,7 @@ export async function runCrawlPhase(
|
|||||||
...params,
|
...params,
|
||||||
chunkNo,
|
chunkNo,
|
||||||
attemptedBefore: attemptedTotal,
|
attemptedBefore: attemptedTotal,
|
||||||
|
startWindow: windowHint,
|
||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
// Apply the chunk's counters even when it did no new work (a retried
|
// Apply the chunk's counters even when it did no new work (a retried
|
||||||
@ -110,6 +126,9 @@ export async function runCrawlPhase(
|
|||||||
// with up-to-date scratchpad totals) — finalize must not see stale ones.
|
// with up-to-date scratchpad totals) — finalize must not see stale ones.
|
||||||
attemptedTotal = result.attempted;
|
attemptedTotal = result.attempted;
|
||||||
pending = result.pending;
|
pending = result.pending;
|
||||||
|
// `?? initial`: an instance in flight across a deploy replays cached
|
||||||
|
// step results from before endWindow existed.
|
||||||
|
windowHint = result.endWindow ?? CRAWL_WINDOW.initial;
|
||||||
// One zero-attempt chunk is normal (retry of a completed chunk number);
|
// One zero-attempt chunk is normal (retry of a completed chunk number);
|
||||||
// two in a row means the frontier is unservable — stop with what we
|
// two in a row means the frontier is unservable — stop with what we
|
||||||
// have instead of spinning forever.
|
// have instead of spinning forever.
|
||||||
@ -122,8 +141,17 @@ export async function runCrawlPhase(
|
|||||||
}
|
}
|
||||||
|
|
||||||
async function runCrawlChunk(
|
async function runCrawlChunk(
|
||||||
input: CrawlPhaseParams & { chunkNo: number; attemptedBefore: number },
|
input: CrawlPhaseParams & {
|
||||||
): Promise<{ attemptedInChunk: number; attempted: number; pending: number }> {
|
chunkNo: number;
|
||||||
|
attemptedBefore: number;
|
||||||
|
startWindow: number;
|
||||||
|
},
|
||||||
|
): Promise<{
|
||||||
|
attemptedInChunk: number;
|
||||||
|
attempted: number;
|
||||||
|
pending: number;
|
||||||
|
endWindow: number;
|
||||||
|
}> {
|
||||||
const { auditId, workflowInstanceId, origin, maxPages, robots, chunkNo } =
|
const { auditId, workflowInstanceId, origin, maxPages, robots, chunkNo } =
|
||||||
input;
|
input;
|
||||||
const scratchpad = getAuditScratchpad(auditId);
|
const scratchpad = getAuditScratchpad(auditId);
|
||||||
@ -132,23 +160,34 @@ async function runCrawlChunk(
|
|||||||
CHUNK_TARGET_PAGES,
|
CHUNK_TARGET_PAGES,
|
||||||
maxPages - input.attemptedBefore,
|
maxPages - input.attemptedBefore,
|
||||||
);
|
);
|
||||||
const claimed = await scratchpad.claimChunk(chunkNo, claimLimit);
|
const { urls: claimed, isRetry } = await scratchpad.claimChunk(
|
||||||
|
chunkNo,
|
||||||
|
claimLimit,
|
||||||
|
);
|
||||||
if (claimed.length === 0) {
|
if (claimed.length === 0) {
|
||||||
const stats = await scratchpad.getStats();
|
const stats = await scratchpad.getStats();
|
||||||
return {
|
return {
|
||||||
attemptedInChunk: 0,
|
attemptedInChunk: 0,
|
||||||
attempted: stats.attempted,
|
attempted: stats.attempted,
|
||||||
pending: stats.pending,
|
pending: stats.pending,
|
||||||
|
endWindow: input.startWindow,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
const depthByUrl = new Map(claimed.map((entry) => [entry.url, entry.depth]));
|
const depthByUrl = new Map(claimed.map((entry) => [entry.url, entry.depth]));
|
||||||
const deadlineAt = Date.now() + CHUNK_SOFT_DEADLINE_MS;
|
const deadlineAt = Date.now() + CHUNK_SOFT_DEADLINE_MS;
|
||||||
|
|
||||||
let windowSize = INITIAL_CRAWL_WINDOW;
|
// A retry means the previous attempt died mid-crawl (in production almost
|
||||||
|
// always exceededMemory), and it is the chunk's last attempt — so it runs
|
||||||
|
// under drastically reduced limits instead of the profile that just failed.
|
||||||
|
const limits = isRetry ? RETRY_CRAWL_WINDOW : CRAWL_WINDOW;
|
||||||
|
let windowSize = isRetry
|
||||||
|
? limits.initial
|
||||||
|
: clampCrawlWindow(input.startWindow, limits);
|
||||||
let nextIndex = 0;
|
let nextIndex = 0;
|
||||||
let attemptedInChunk = 0;
|
let attemptedInChunk = 0;
|
||||||
const inFlight = new Set<Promise<void>>();
|
const inFlight = new Set<Promise<void>>();
|
||||||
|
let persistThreshold = FIRST_PERSIST_BATCH_SIZE;
|
||||||
let batch: CrawledPageResult[] = [];
|
let batch: CrawledPageResult[] = [];
|
||||||
// Persistence runs concurrently with fetching (pipelined) but sequentially
|
// Persistence runs concurrently with fetching (pipelined) but sequentially
|
||||||
// with itself, so DB write pressure stays bounded at one batch at a time.
|
// with itself, so DB write pressure stays bounded at one batch at a time.
|
||||||
@ -159,7 +198,8 @@ async function runCrawlChunk(
|
|||||||
if (batch.length === 0) return;
|
if (batch.length === 0) return;
|
||||||
const pages = batch;
|
const pages = batch;
|
||||||
batch = [];
|
batch = [];
|
||||||
windowSize = adjustCrawlWindow(windowSize, pages);
|
persistThreshold = PERSIST_BATCH_SIZE;
|
||||||
|
windowSize = adjustCrawlWindow(windowSize, pages, limits);
|
||||||
queuedPersists += 1;
|
queuedPersists += 1;
|
||||||
persistChain = persistChain
|
persistChain = persistChain
|
||||||
.then(() =>
|
.then(() =>
|
||||||
@ -184,7 +224,7 @@ async function runCrawlChunk(
|
|||||||
.then((page) => {
|
.then((page) => {
|
||||||
attemptedInChunk += 1;
|
attemptedInChunk += 1;
|
||||||
batch.push(page);
|
batch.push(page);
|
||||||
if (batch.length >= PERSIST_BATCH_SIZE) flush();
|
if (batch.length >= persistThreshold) flush();
|
||||||
})
|
})
|
||||||
.finally(() => {
|
.finally(() => {
|
||||||
inFlight.delete(promise);
|
inFlight.delete(promise);
|
||||||
@ -239,6 +279,7 @@ async function runCrawlChunk(
|
|||||||
attemptedInChunk,
|
attemptedInChunk,
|
||||||
attempted: stats.attempted,
|
attempted: stats.attempted,
|
||||||
pending: stats.pending,
|
pending: stats.pending,
|
||||||
|
endWindow: windowSize,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user