crawler: per-run time budget (RI_MAX_RUN_MINUTES) so large sources do not monopolise crawl slots

Simon-Pierre Boucher committed 17 days ago (Sep 7, 2026) parent ee3f29a

1 changed file +12 −2

modified workers/crawler/run.ts +12 −2

		`@@ -88,8 +88,16 @@ export async function runCrawl(connectorId: string, options: Partial<CrawlOption`
88	88	if (options.onRaw && inserted.length) await options.onRaw(inserted.map((r) => r.id));
89	89	}
90	90
	91	+ // Time budget per run: very large sources (100k+ pages) would otherwise monopolise crawl slots for hours.
	92	+ // The connector's cursor is persisted at its checkpoints, so a time-boxed run simply resumes on the next tick.
	93	+ const maxRunMs = Number(process.env.RI_MAX_RUN_MINUTES ?? 40) * 60_000;
	94	+ let timeBoxed = false;
91	95	try {
92	96	for await (const rec of connector.crawl(ctx)) {
	97	+ if (mode !== 'probe' && Date.now() - started > maxRunMs) {
	98	+ timeBoxed = true;
	99	+ break;
	100	+ }
93	101	const input: RawRecordInput = rec;
94	102	const payloadText = typeof input.payload === 'string' ? input.payload : JSON.stringify(input.payload ?? null);
95	103	const contentHash = sha256(`${input.kind}\|${input.externalId ?? input.url}\|${payloadText}`);
		`@@ -137,14 +145,16 @@ export async function runCrawl(connectorId: string, options: Partial<CrawlOption`
137	145	const dupAnomaly = duplicateExplosion(raw + dupes, dupes);
138	146	if (dupAnomaly && mode !== 'backfill') anomalies.push(dupAnomaly);
139	147	if (attempted > 0 && success === 0) anomalies.push('all_pages_failed');
140		− const status: CrawlRunResult['status'] = error ? (raw > 0 ? 'partial' : 'failed') : 'success';
	148	+ if (timeBoxed) anomalies.push('time_budget_reached');
	149	+ const status: CrawlRunResult['status'] = error ? (raw > 0 ? 'partial' : 'failed') : timeBoxed ? 'partial' : 'success';
141	150	const finishedAt = new Date();
142	151	await db()
143	152	.update(connectorRuns)
144	153	.set({ finishedAt, status, pagesAttempted: attempted, pagesSuccess: success, recordsRaw: raw, recordsDuplicate: dupes, engineStats: stats, anomalies: [...anomalies, ...ctx.anomalies.filter((a) => !anomalies.includes(a))], error, costCredits: credits, cursor: cursor ?? savedCursor ?? {} })
145	154	.where(eq(connectorRuns.id, runId));
146	155	const refresh = state?.refreshFrequencyMinutes ?? meta.refreshFrequencyMinutes;
147		− const nextRunAt = new Date(finishedAt.getTime() + refresh * 60_000 * (status === 'failed' ? 2 : 1));
	156	+ // A time-boxed run resumes quickly (cursor kept); failures back off ×2.
	157	+ const nextRunAt = timeBoxed ? new Date(finishedAt.getTime() + 3 * 60_000) : new Date(finishedAt.getTime() + refresh * 60_000 * (status === 'failed' ? 2 : 1));
148	158	await db()
149	159	.update(connectorsTable)
150	160	.set({
151	161