From 07d8568f6c8e6b8d5cb54b3411ba17995f0f368a Mon Sep 17 00:00:00 2001 From: Jan Buchar Date: Mon, 7 Sep 2026 15:57:45 +0200 Subject: [PATCH 1/3] refactor: Extract per-run crawler state into a CrawlerRun class --- docs/public-api/crawlee-basic.api.md | 6 +- docs/public-api/crawlee-browser-pool.api.md | 2 + docs/public-api/crawlee-browser.api.md | 4 + docs/public-api/crawlee-core.api.md | 15 +- docs/public-api/crawlee-playwright.api.md | 2 + docs/public-api/crawlee-types.api.md | 17 +- docs/upgrading/upgrading_v4.md | 23 ++ .../src/internals/basic-crawler.ts | 230 +++++++++--------- .../src/internals/crawler-run.ts | 180 ++++++++++++++ .../src/internals/browser-crawler.ts | 21 +- packages/browser-pool/src/browser-pool.ts | 25 +- .../browser-pool/src/remote-browser-pool.ts | 8 + packages/core/src/log.ts | 53 +++- .../internals/adaptive-playwright-crawler.ts | 25 +- packages/types/src/logger.ts | 22 +- test/core/crawlers/basic_crawler.test.ts | 46 +++- test/core/crawlers/browser_crawler.test.ts | 60 +++-- 17 files changed, 545 insertions(+), 194 deletions(-) create mode 100644 packages/basic-crawler/src/internals/crawler-run.ts diff --git a/docs/public-api/crawlee-basic.api.md b/docs/public-api/crawlee-basic.api.md index f5dd7d1d987a..a327bb429eab 100644 --- a/docs/public-api/crawlee-basic.api.md +++ b/docs/public-api/crawlee-basic.api.md @@ -47,6 +47,8 @@ import type { TypedRequestsLike } from '@crawlee/core'; // @public (undocumented) export class BasicCrawler, ExtendedContext extends Context = Context & ContextExtension, Routes extends Record = Record>, StatisticStateExtension extends object = {}> { + // (undocumented) + [Symbol.asyncDispose](): Promise; constructor(options?: BasicCrawlerOptions & RequireContextPipeline); // (undocumented) protected readonly additionalHttpErrorStatusCodes: Set; @@ -61,6 +63,7 @@ export class BasicCrawler; protected createDefaultConcurrencySystem(options: ConcurrencySystemOptions): ConcurrencySystem; + destroy(): Promise; exportData(path: string, format?: 'json' | 'csv', options?: DatasetExportOptions): Promise; // (undocumented) protected getCookieHeaderFromRequest(request: Request_2): string; @@ -95,8 +98,7 @@ export class BasicCrawler; run(requests?: TypedRequestsLike, options?: CrawlerRunOptions): Promise; - // (undocumented) - running: boolean; + get running(): boolean; // (undocumented) protected runRequestHandler(crawlingContext: ExtendedContext): Promise; get sessionPool(): ISessionPool; diff --git a/docs/public-api/crawlee-browser-pool.api.md b/docs/public-api/crawlee-browser-pool.api.md index 8c42792af2bd..3938321c202d 100644 --- a/docs/public-api/crawlee-browser-pool.api.md +++ b/docs/public-api/crawlee-browser-pool.api.md @@ -229,6 +229,7 @@ export class BrowserPool[]; // (undocumented) prePageCreateHooks: PrePageCreateHook[]; + releaseAllBrowsers(): Promise; retireAllBrowsers(): void; // (undocumented) retireBrowserAfterPageCount: number; @@ -587,6 +588,7 @@ export class RemoteBrowserPool implements IBrowserPool { get maxOpenBrowsers(): number; set maxOpenBrowsers(value: number); newPage(options?: NewPageOptions): Promise; + releaseAllBrowsers(): Promise; } // @public (undocumented) diff --git a/docs/public-api/crawlee-browser.api.md b/docs/public-api/crawlee-browser.api.md index ab664946a277..140e0132ebc4 100644 --- a/docs/public-api/crawlee-browser.api.md +++ b/docs/public-api/crawlee-browser.api.md @@ -48,6 +48,8 @@ export abstract class BrowserCrawler>; // (undocumented) + destroy(): Promise; + // (undocumented) protected getNavigationTimeoutMillis(): number; // (undocumented) protected readonly ignoreIframes: boolean; @@ -58,6 +60,7 @@ export abstract class BrowserCrawler, gotoOptions: GoToOptions): Promise; protected runRequestHandler(crawlingContext: ExtendedContext): Promise; + teardown(): Promise; } // @public (undocumented) @@ -113,6 +116,7 @@ export type LauncherRemoteBrowserPoolOptions = Omit = IBrowserPool & { + releaseAllBrowsers: () => Promise; destroy: () => Promise; }; diff --git a/docs/public-api/crawlee-core.api.md b/docs/public-api/crawlee-core.api.md index 0a842f6878d2..b463b26afdf5 100644 --- a/docs/public-api/crawlee-core.api.md +++ b/docs/public-api/crawlee-core.api.md @@ -32,6 +32,7 @@ import { LoggerJson } from '@apify/log'; import type { LoggerOptions } from '@apify/log'; import { LoggerText } from '@apify/log'; import { LogLevel } from '@apify/log'; +import type { LogOptions } from '@crawlee/types'; import { ParseSitemapOptions } from '@crawlee/utils'; import type { ProcessedRequest } from '@crawlee/types'; import type { ProxyInfo } from '@crawlee/types'; @@ -90,26 +91,26 @@ export abstract class BaseCrawleeLogger implements CrawleeLogger { child(options: Partial): CrawleeLogger; protected abstract createChild(options: Partial): CrawleeLogger; // (undocumented) - debug(message: string, data?: Record): void; + debug(message: string, data?: Record, options?: LogOptions): void; // (undocumented) deprecated(message: string): void; // (undocumented) - error(message: string, data?: Record): void; + error(message: string, data?: Record, options?: LogOptions): void; // (undocumented) exception(exception: Error, message: string, data?: Record): void; // (undocumented) getOptions(): CrawleeLoggerOptions; // (undocumented) - info(message: string, data?: Record): void; + info(message: string, data?: Record, options?: LogOptions): void; abstract logWithLevel(level: number, message: string, data?: Record): void; // (undocumented) - perf(message: string, data?: Record): void; + perf(message: string, data?: Record, options?: LogOptions): void; // (undocumented) setOptions(options: Partial): void; // (undocumented) - softFail(message: string, data?: Record): void; + softFail(message: string, data?: Record, options?: LogOptions): void; // (undocumented) - warning(message: string, data?: Record): void; + warning(message: string, data?: Record, options?: LogOptions): void; // (undocumented) warningOnce(message: string): void; } @@ -966,6 +967,8 @@ export { LoggerText } export { LogLevel } +export { LogOptions } + // @public (undocumented) export const MAX_POOL_SIZE = 1000; diff --git a/docs/public-api/crawlee-playwright.api.md b/docs/public-api/crawlee-playwright.api.md index d1e3ed444b6f..1bfa7d7d996b 100644 --- a/docs/public-api/crawlee-playwright.api.md +++ b/docs/public-api/crawlee-playwright.api.md @@ -75,6 +75,8 @@ export class AdaptivePlaywrightCrawler, Ext // (undocumented) protected buildContextPipeline(): ContextPipeline_2; // (undocumented) + destroy(): Promise; + // (undocumented) protected init(): Promise; // (undocumented) protected runRequestHandler(crawlingContext: CrawlingContext_2): Promise; diff --git a/docs/public-api/crawlee-types.api.md b/docs/public-api/crawlee-types.api.md index 04b8e449cd20..8b1b7ad43f75 100644 --- a/docs/public-api/crawlee-types.api.md +++ b/docs/public-api/crawlee-types.api.md @@ -96,17 +96,17 @@ export interface CookieJarSetCookieOptions { // @public export interface CrawleeLogger { child(options: Partial): CrawleeLogger; - debug(message: string, data?: Record): void; + debug(message: string, data?: Record, options?: LogOptions): void; deprecated(message: string): void; - error(message: string, data?: Record): void; + error(message: string, data?: Record, options?: LogOptions): void; exception(exception: Error, message: string, data?: Record): void; getOptions(): CrawleeLoggerOptions; - info(message: string, data?: Record): void; + info(message: string, data?: Record, options?: LogOptions): void; logWithLevel(level: number, message: string, data?: Record): void; - perf(message: string, data?: Record): void; + perf(message: string, data?: Record, options?: LogOptions): void; setOptions(options: Partial): void; - softFail(message: string, data?: Record): void; - warning(message: string, data?: Record): void; + softFail(message: string, data?: Record, options?: LogOptions): void; + warning(message: string, data?: Record, options?: LogOptions): void; warningOnce(message: string): void; } @@ -309,6 +309,11 @@ export interface KeyValueStoreRecord { // @public export type KeyValueStoreRecordInputValue = Buffer | ArrayBuffer | ArrayBufferView | string | NodeJS.ReadableStream | ReadableStream; +// @public +export interface LogOptions { + once?: boolean; +} + // @public export interface NewPageOptions { id?: string; diff --git a/docs/upgrading/upgrading_v4.md b/docs/upgrading/upgrading_v4.md index bef92b89cba8..6243a05664fc 100644 --- a/docs/upgrading/upgrading_v4.md +++ b/docs/upgrading/upgrading_v4.md @@ -652,6 +652,29 @@ for (const [index, urls] of batches.entries()) { An alias identifies a run-scoped queue. It has no persistent name, and is emptied on start along with the default storages. Reuse an alias and you get that same queue back, handled requests included. The next crawl then finds nothing to do. Give each crawl its own alias. `purge()` is for when one crawler and one queue must be reused. +### `teardown()` is per-run, disposing of the crawler is not + +`crawler.teardown()` ends the run in progress and releases only what that run owns — it is what `run()` calls on its way out. In v3 it also destroyed the browser pool a browser crawler had built for itself, and a destroyed `BrowserPool` cannot be used again: with its timers cleared and its listeners dropped, a second `run()` had nothing retiring idle browsers or reaping the retired ones. It now releases that run's browsers and leaves the pool usable. + +What outlives a run is released by `crawler.destroy()`, or by disposing of the crawler: + +```typescript +await using crawler = new PlaywrightCrawler({ requestHandler: async ({ page }) => { /* ... */ } }); + +await crawler.run(['https://example.com/a']); +await crawler.run(['https://example.com/b']); +``` + +Disposing is optional — a finished run leaves no browsers open and no timer holding the process alive. + +:::info + +The `await using` syntax needs Node.js 24 or later. On Node.js 22 call `destroy()` yourself instead — it is what the disposal hook calls anyway, as with the [collaborators you own](#collaborators-you-own-are-disposable). + +::: + +`crawler.running` is now a read-only getter; in v3 it was an assignable field. + ### Storage `.open()` now also accepts `{ id?, name? }` `Dataset.open()`, `KeyValueStore.open()`, and `RequestQueue.open()` previously accepted a single `idOrName?: string` parameter. This was ambiguous — callers couldn't express whether they were opening a storage by its ID or by name. diff --git a/packages/basic-crawler/src/internals/basic-crawler.ts b/packages/basic-crawler/src/internals/basic-crawler.ts index 7ba2f8c606bf..42c9043c4a1f 100644 --- a/packages/basic-crawler/src/internals/basic-crawler.ts +++ b/packages/basic-crawler/src/internals/basic-crawler.ts @@ -34,7 +34,6 @@ import type { } from '@crawlee/core'; import { applyRequestTransform, - AutoscaledPool, bindMethodsToServiceLocator, BLOCKED_STATUS_CODES, buildEnqueueStrategyPatterns, @@ -105,6 +104,7 @@ import { LruCache } from '@apify/datastructures'; import { addTimeoutToPromise, extendTimeout, storage as timeoutStorage, TimeoutError, tryCancel } from '@apify/timeout'; import { cryptoRandomObjectId } from '@apify/utilities'; +import { CrawlerRun } from './crawler-run.js'; import { extendTimeoutKey, navigationDeadlineKey, @@ -765,40 +765,42 @@ export class BasicCrawler< */ readonly #resolveConcurrencySystem: () => OwnedOrInjected; - /** As resolved by `init()`. Absent until the first run, so a `teardown()` before it is a no-op. */ - #concurrencySystemDep?: OwnedOrInjected; + /** The last run started on this instance. It outlives the run, so anything that drives one goes through `#liveRun`. */ + #run?: CrawlerRun; + + #concurrencySystem: OwnedOrInjected | undefined; + + /** The run in progress, if any — a finished run is history and cannot be driven. */ + get #liveRun(): CrawlerRun | undefined { + return this.#run?.isLive === true ? this.#run : undefined; + } + + /** The run whose task loop is dispatching, if any — only that one can be paused and resumed. */ + get #dispatchingRun(): CrawlerRun | undefined { + return this.#liveRun?.isDispatching === true ? this.#liveRun : undefined; + } + + /** Whether a {@apilink BasicCrawler.run|`run()`} is in progress on this instance. */ + get running(): boolean { + return this.#liveRun !== undefined; + } /** - * The concurrency governor this run is booking its requests against — either the + * The concurrency governor the run in progress is booking its requests against — either the * {@apilink BasicCrawlerOptions.concurrencySystem|`concurrencySystem`} that was injected, or the default the * crawler built for itself. Read it for telemetry: `desiredConcurrency`, `currentConcurrency`, `isRunning`. * - * > *NOTE:* `undefined` until {@apilink BasicCrawler.run|`crawler.run()`} has resolved it. A crawler-owned default - * is also rebuilt for every run, so the instance is not stable across runs. + * > *NOTE:* `undefined` outside a run, and a crawler-owned default is rebuilt for every run — so read it + * during a run rather than caching it across runs. * * {@apilink IConcurrencySystem} is deliberately read-only. Tuning concurrency *while a crawl is running* means * owning the instance: build a {@apilink ConcurrencySystem} yourself and inject it, then set * `minConcurrency`/`maxConcurrency`/`desiredConcurrency` on your own reference. */ get concurrencySystem(): IConcurrencySystem | undefined { - return this.#concurrencySystemDep?.maybeValue; + return this.#concurrencySystem?.value; } - /** - * The task loop that dispatches this run's requests. Private on purpose — it is a bare parallel task runner with - * no configuration left of its own (see {@apilink ConcurrencySystem}), and everything a caller legitimately did - * with it now has a crawler-level counterpart: {@apilink BasicCrawler.pause|`pause()`}, - * {@apilink BasicCrawler.resume|`resume()`}, {@apilink BasicCrawler.teardown|`teardown()`} and - * {@apilink BasicCrawler.concurrencySystem|`concurrencySystem`}. - */ - #autoscaledPool?: AutoscaledPool; - - /** A pending nudge of the task loop, armed when the request manager announces when it will have work again. */ - #taskLoopWakeTimer?: NodeJS.Timeout; - - /** When the pending wake-up is due, so an earlier one can replace a later one. */ - #taskLoopWakeAt = 0; - /** * A reference to the underlying {@apilink IProxyConfiguration} instance that manages the crawler's proxies. * Only available if used by the crawler. @@ -842,9 +844,7 @@ export class BasicCrawler< return this.#contextPipeline; } - running = false; #hasFinishedBefore = false; - #unexpectedStop = false; /** Whether a `run()` on this instance has already finished - a repeated one continues where it left off. */ get hasFinishedBefore(): boolean { @@ -890,8 +890,6 @@ export class BasicCrawler< /** The resolved per-storage-type write policy overrides forwarded to each request's transaction. */ readonly #storageWritePolicy: Partial; readonly #onSkippedRequest?: SkippedRequestCallback; - #closeEvents?: boolean; - #loggedPerRun = new Set(); readonly #robotsTxtFileCache: LruCache; readonly #identity: CrawlerIdentity; readonly #contextPipelineOptions: { @@ -1316,7 +1314,7 @@ export class BasicCrawler< }, isTaskReadyFunction: async () => { if (isMaxPagesExceeded()) { - this.logOncePerRun( + this.#liveRun?.logOnce( 'shuttingDown', 'Crawler reached the maxRequestsPerCrawl limit of ' + `${this.#maxRequestsPerCrawl} requests and will shut down soon. Requests that are in progress will be allowed to finish.`, @@ -1324,8 +1322,8 @@ export class BasicCrawler< return false; } - if (this.#unexpectedStop) { - this.logOncePerRun( + if (this.#run?.stopRequested) { + this.#liveRun?.logOnce( 'shuttingDown', 'No new requests are allowed because the `stop()` method has been called. ' + 'Ongoing requests will be allowed to complete.', @@ -1345,7 +1343,7 @@ export class BasicCrawler< return true; } - if (this.#unexpectedStop) { + if (this.#run?.stopRequested) { this.log.info( 'The crawler has finished all the remaining ongoing requests and will shut down now.', ); @@ -1712,9 +1710,18 @@ export class BasicCrawler< }); } - this.#unexpectedStop = false; - this.running = true; - this.#loggedPerRun.clear(); + // An owned governor is rebuilt for every run, so it always starts from a clean slate — stale + // resource snapshots or a previous run's scaled desired concurrency would otherwise distort this + // run's scaling. An injected one is long-lived and its lifecycle belongs to the caller. + this.#concurrencySystem = this.#resolveConcurrencySystem(); + const run = new CrawlerRun({ + log: this.log, + concurrencySystem: this.#concurrencySystem.value, + taskLoopOptions: this.#taskLoopOptions, + consumer: this.#identity, + }); + + this.#run = run; await purgeDefaultStorages({ onlyPurgeOnce: true, @@ -1727,6 +1734,9 @@ export class BasicCrawler< } try { + // Inside the try: from here on, a failed startup has something to stop again. An injected + // governor is the caller's to run, and the task loop rejects an unstarted one. + await this.#concurrencySystem.ifOwned((system) => system.start()); await this.init(); await this.statistics.startCapturing(); } catch (error) { @@ -1739,7 +1749,8 @@ export class BasicCrawler< }); // The run never began, so let the instance be run again instead of leaving it wedged as `running`. - this.running = false; + await run.finish(); + await this.#concurrencySystem.ifOwned((system) => system.stop()); throw error; } @@ -1751,7 +1762,7 @@ export class BasicCrawler< 'Pausing... Press CTRL+C again to force exit. To resume, do: CRAWLEE_PURGE_ON_START=0 npm start', ); await this.pauseOnMigration(); - await this.#autoscaledPool!.abort(); + await run.abort(); }; // Attach a listener to handle migration and aborting events gracefully. @@ -1764,7 +1775,7 @@ export class BasicCrawler< let stats = {} as FinalStatistics; try { - await this.#autoscaledPool!.run(); + await run.dispatchRequests(); } finally { await this.statistics.stopCapturing(); await this.teardown(); @@ -1834,7 +1845,8 @@ export class BasicCrawler< { isStatusMessageTerminal: true, level: 'INFO' }, ); - this.running = false; + await run.finish(); + await this.#concurrencySystem.ifOwned((system) => system.stop()); this.#hasFinishedBefore = true; } @@ -1851,11 +1863,14 @@ export class BasicCrawler< * To stop the crawler immediately, use {@apilink BasicCrawler.teardown|`crawler.teardown()`} instead. */ stop(reason = 'The crawler has been gracefully stopped.'): void { - if (this.#unexpectedStop) { + const run = this.#liveRun; + + if (run === undefined) { + this.log.warning('Cannot stop a crawler that is not running.'); return; } - this.log.info(reason); - this.#unexpectedStop = true; + + run.stop(reason); } /** @@ -1867,12 +1882,14 @@ export class BasicCrawler< * throughout, since a shared one may still be serving other crawlers. */ async pause(timeoutSecs?: number): Promise { - if (!this.#autoscaledPool) { + const run = this.#dispatchingRun; + + if (run === undefined) { this.log.warning('Cannot pause a crawler that is not running.'); return; } - await this.#autoscaledPool.pause(timeoutSecs); + await run.pause(timeoutSecs); } /** @@ -1880,12 +1897,14 @@ export class BasicCrawler< * again. A no-op on a crawler that is not paused. */ resume(): void { - if (!this.#autoscaledPool) { + const run = this.#dispatchingRun; + + if (run === undefined) { this.log.warning('Cannot resume a crawler that is not running.'); return; } - this.#autoscaledPool.resume(); + run.resume(); } /** @@ -2022,18 +2041,22 @@ export class BasicCrawler< // A skipped request is a *successful* outcome, but the interrupt still unwinds through the // transaction scope, which rolls back - so the skip bookkeeping must write directly. await withDirectStorageAccess(async () => { + // Enqueueing happens outside a run too - `addRequests()` before the first `run()` - so these dedupe + // against the logger for the crawler's lifetime rather than per run. if (options.reason === 'limit') { - this.logOncePerRun( - 'maxRequestsPerCrawl', + this.log.info( 'The number of requests enqueued by the crawler reached the maxRequestsPerCrawl limit of ' + `${this.#maxRequestsPerCrawl} requests and no further requests will be added.`, + undefined, + { once: true }, ); } if (options.reason === 'depth') { - this.logOncePerRun( - 'maxCrawlDepth', + this.log.info( `The crawler reached the maxCrawlDepth limit of ${this.#maxCrawlDepth} and no further requests will be enqueued.`, + undefined, + { once: true }, ); } @@ -2041,13 +2064,6 @@ export class BasicCrawler< }); } - private logOncePerRun(key: string, message: string, level: 'info' | 'warning' = 'info'): void { - if (!this.#loggedPerRun.has(key)) { - this.log[level](message); - this.#loggedPerRun.add(key); - } - } - /** * Adds requests to the queue in batches. By default, it will resolve after the initial batch is added, and continue * adding the rest in background. You can configure the batch size via `batchSize` option and the sleep time in between @@ -2328,11 +2344,17 @@ export class BasicCrawler< * Initializes the crawler. */ protected async init(): Promise { + const run = this.#liveRun; + + if (run === undefined) { + throw new Error('There is no crawl in progress - `run()` is the only thing that starts one.'); + } + const eventManager = serviceLocator.getEventManager(); if (!eventManager.isInitialized()) { await eventManager.init(); - this.#closeEvents = true; + run.ownsEventManager = true; } // Warn once at startup if the internal timeout is shorter than the phases it is meant to outlast. It is @@ -2348,18 +2370,6 @@ export class BasicCrawler< ); } - // An owned governor is rebuilt (and started) for every run, so it always starts from a clean slate — stale - // resource snapshots or a previous run's scaled desired concurrency would otherwise distort this run's - // scaling. An injected one is long-lived and its lifecycle belongs to the caller. - this.#concurrencySystemDep = this.#resolveConcurrencySystem(); - await this.#concurrencySystemDep.ifOwned((system) => system.start()); - - this.#autoscaledPool = new AutoscaledPool({ - ...this.#taskLoopOptions, - concurrencySystem: this.#concurrencySystemDep.value, - consumer: this.#identity, - }); - await this.getRequestManager(); } @@ -2514,13 +2524,11 @@ export class BasicCrawler< } const domain = hostnameOrUrl(url); - this.logOncePerRun( - `rateLimitNotThrottled:${domain}`, + this.log.warningOnce( `"${domain}" responded with HTTP 429 (Too Many Requests), but the crawler's request manager does not ` + 'pace that domain, so the response is handled like any other, with no per-domain delay. Set ' + `\`sameDomainDelaySecs\`, or pass a \`ThrottlingRequestManager\` covering "${domain}" as ` + '`requestManager`, to honour `Retry-After` and apply exponential backoff instead.', - 'warning', ); return false; @@ -2542,13 +2550,11 @@ export class BasicCrawler< } const domain = hostnameOrUrl(url); - this.logOncePerRun( - `crawlDelayIgnored:${domain}`, + this.log.warningOnce( `robots.txt for "${domain}" defines a crawl-delay of ${delaySeconds}s, but the crawler's request ` + 'manager does not pace that domain, so its requests will not be paced. Set ' + `\`sameDomainDelaySecs\`, or pass a \`ThrottlingRequestManager\` covering "${domain}" as ` + '`requestManager`.', - 'warning', ); } @@ -2576,9 +2582,10 @@ export class BasicCrawler< } private async pauseOnMigration() { - if (this.#autoscaledPool) { - // if run wasn't called, this is going to crash - await this.#autoscaledPool.pause(SAFE_MIGRATION_WAIT_MILLIS).catch((err) => { + const run = this.#dispatchingRun; + + if (run) { + await run.pause(SAFE_MIGRATION_WAIT_MILLIS).catch((err) => { if (err.message.includes('running tasks did not finish')) { this.log.error( 'The crawler was paused due to migration to another host, ' + @@ -2778,44 +2785,12 @@ export class BasicCrawler< const state = await this.requestManager.checkReadiness(); if (state.status === 'waiting' && state.readyAt !== undefined) { - this.#scheduleTaskLoopWake(state.readyAt); + this.#run?.scheduleWake(state.readyAt); } return state.status === 'ready'; } - /** - * Nudges the task loop at `readyAt`, on a single timer that only an earlier one replaces. The pool polls anyway - * every `maybeRunIntervalSecs` (0.5s by default), so this only shortens the wait - hence one timer rather than - * one per probe, and `unref`'d so it never keeps the process alive. - */ - #scheduleTaskLoopWake(readyAt: number): void { - if (this.#taskLoopWakeTimer !== undefined) { - if (this.#taskLoopWakeAt <= readyAt) { - return; - } - clearTimeout(this.#taskLoopWakeTimer); - } - - this.#taskLoopWakeAt = readyAt; - this.#taskLoopWakeTimer = setTimeout( - () => { - this.#taskLoopWakeTimer = undefined; - void this.#autoscaledPool?.notify(); - }, - Math.max(0, readyAt - Date.now()), - ); - this.#taskLoopWakeTimer.unref(); - } - - /** Drops a pending task-loop wake-up, so a finished run leaves no timer behind. */ - #clearTaskLoopWake(): void { - if (this.#taskLoopWakeTimer !== undefined) { - clearTimeout(this.#taskLoopWakeTimer); - this.#taskLoopWakeTimer = undefined; - } - } - /** * Unwraps errors thrown by the context pipeline to get the actual user error. * RequestHandlerError and ContextPipelineInitializationError wrap the actual error. @@ -2978,13 +2953,18 @@ export class BasicCrawler< } /** - * Stops the crawler immediately. + * Ends the run in progress immediately, without waiting for the requests already in flight. * - * This method doesn't wait for currently active requests to finish. + * This runs at the end of every {@apilink BasicCrawler.run|`run()`}, so what it releases is per-run only, and + * a repeated `run()` has to find everything it needs either intact or re-establishable. What outlives a run is + * released by {@apilink BasicCrawler.destroy|`destroy()`} instead. * * To stop the crawler gracefully (waiting for all running requests to finish), use {@apilink BasicCrawler.stop|`crawler.stop()`} instead. */ async teardown(): Promise { + const eventManager = serviceLocator.getEventManager(); + const ownsEventManager = this.#run?.ownsEventManager === true; + // First, so that the pool writes its final state once - its teardown stops it listening before the // persistence event below goes out. await this.#sessionPoolDep.ifOwned(async (pool) => pool.teardown()); @@ -2992,17 +2972,29 @@ export class BasicCrawler< // When this crawler initialized the event manager, its close() call emits // the final persistence event after the crawler-specific state has been // saved. External event managers still need an explicit event here. - if (!this.#closeEvents) { - serviceLocator.getEventManager().emit(EventType.PERSIST_STATE, { isMigrating: false }); + if (!ownsEventManager) { + eventManager.emit(EventType.PERSIST_STATE, { isMigrating: false }); } - if (this.#closeEvents) { - await serviceLocator.getEventManager().close(); + if (ownsEventManager) { + await eventManager.close(); } - this.#clearTaskLoopWake(); - await this.#autoscaledPool?.abort(); - await this.#concurrencySystemDep?.ifOwned((system) => system.stop()); + await this.#run?.end(); + } + + /** + * Releases what the crawler owns beyond a single run — in the browser crawlers, the browser pool. A finished + * `run()` has already released everything a run owns, so this is only needed before dropping a crawler you + * could otherwise have run again. + */ + async destroy(): Promise { + // An abandoned run - one whose `run()` never returned - would otherwise keep dispatching requests. + await this.#run?.end(); + } + + async [Symbol.asyncDispose](): Promise { + await this.destroy(); } protected getCookieHeaderFromRequest(request: Request) { diff --git a/packages/basic-crawler/src/internals/crawler-run.ts b/packages/basic-crawler/src/internals/crawler-run.ts new file mode 100644 index 000000000000..ad8085d49c2a --- /dev/null +++ b/packages/basic-crawler/src/internals/crawler-run.ts @@ -0,0 +1,180 @@ +import type { AutoscaledPoolOptions, CrawleeLogger, IConcurrencySystem } from '@crawlee/core'; +import { AutoscaledPool } from '@crawlee/core'; + +/** Everything a run needs to exist. Whatever a crawler resolves per run is resolved before the run is built. */ +export interface CrawlerRunSetup { + log: CrawleeLogger; + /** The governor this run books its requests against, started and stopped by whoever owns it. */ + concurrencySystem: IConcurrencySystem; + taskLoopOptions: Omit; + consumer: AutoscaledPoolOptions['consumer']; +} + +/** + * Everything that belongs to a single {@apilink BasicCrawler.run|`crawler.run()`} — the task loop dispatching its + * requests, the concurrency governor those requests are booked against, and the bookkeeping that must not survive + * into a later run. + * + * The crawler builds a fresh one for every run, which is what keeps per-run state from leaking across runs: there + * is no reset checklist to keep in sync. A finished run stays readable as the history of the last crawl, so + * liveness is asked about explicitly (`isLive`) rather than inferred from presence. + * + * The lifecycle is `dispatchRequests()` and `finish()`. Everything a run owns exists from construction, so there + * is no half-built state to guard against in between. + * + * @internal + */ +export class CrawlerRun { + readonly #log: CrawleeLogger; + readonly #autoscaledPool: AutoscaledPool; + readonly #loggedOnce = new Set(); + + #stopRequested = false; + #dispatching = false; + #ended = false; + #finished = false; + #wakeTimer?: NodeJS.Timeout; + #wakeAt = 0; + + /** Whether this run initialized the ambient event manager, and is therefore the one that closes it. */ + ownsEventManager = false; + + constructor(setup: CrawlerRunSetup) { + this.#log = setup.log; + + this.#autoscaledPool = new AutoscaledPool({ + ...setup.taskLoopOptions, + concurrencySystem: setup.concurrencySystem, + consumer: setup.consumer, + }); + } + + /** + * Runs the task loop until the crawl is over or {@apilink CrawlerRun.abort|aborted}. The governor has to be + * running by now — the task loop refuses to book anything against one that is not. + */ + async dispatchRequests(): Promise { + this.#dispatching = true; + + try { + await this.#autoscaledPool.run(); + } finally { + this.#dispatching = false; + } + } + + /** Whether the `run()` call this belongs to is still executing. */ + get isLive(): boolean { + return !this.#finished; + } + + /** + * Whether the task loop is dispatching requests. Pausing one that has not started would not suspend a crawl, + * it would keep it from ever starting. + */ + get isDispatching(): boolean { + return this.#dispatching; + } + + /** Whether a graceful shutdown of this run has been requested. */ + get stopRequested(): boolean { + return this.#stopRequested; + } + + /** Asks the run to stop taking new requests. Only the first call logs `reason`. */ + stop(reason: string): void { + if (this.#stopRequested) { + return; + } + + this.#log.info(reason); + this.#stopRequested = true; + } + + /** Stops dispatching new requests, resolving once the ones in flight have settled. */ + async pause(timeoutSecs?: number): Promise { + await this.#autoscaledPool.pause(timeoutSecs); + } + + /** Resumes dispatching after a {@apilink CrawlerRun.pause|`pause()`}. */ + resume(): void { + this.#autoscaledPool.resume(); + } + + /** Ends the crawl without waiting for the requests in flight, so a pending `crawl()` resolves. */ + async abort(): Promise { + await this.#autoscaledPool.abort(); + } + + /** + * Logs `message` the first time this run asks for it under `key`. Keyed rather than deduped by text, so that + * two messages describing the same shutdown say it once between them. + */ + logOnce(key: string, message: string): void { + if (this.#loggedOnce.has(key)) { + return; + } + + this.#log.info(message); + this.#loggedOnce.add(key); + } + + /** + * Nudges the task loop at `readyAt`, when the request manager says it will have work again. The loop polls on + * its own anyway, so this only shortens the wait — hence a single timer that only an earlier wake-up replaces, + * `unref`'d so it never keeps the process alive. A run that has ended arms nothing. + */ + scheduleWake(readyAt: number): void { + if (this.#ended) { + return; + } + + if (this.#wakeTimer !== undefined) { + if (this.#wakeAt <= readyAt) { + return; + } + + clearTimeout(this.#wakeTimer); + } + + this.#wakeAt = readyAt; + this.#wakeTimer = setTimeout( + () => { + this.#wakeTimer = undefined; + void this.#autoscaledPool.notify(); + }, + Math.max(0, readyAt - Date.now()), + ); + this.#wakeTimer.unref(); + } + + /** + * Ends the run: drops a pending wake-up and aborts the task loop. Idempotent, so aborting a run through + * `teardown()` and then finishing it does the work once. The governor is left alone — it belongs to whoever + * started it. + */ + async end(): Promise { + if (this.#ended) { + return; + } + + this.#ended = true; + + if (this.#wakeTimer !== undefined) { + clearTimeout(this.#wakeTimer); + this.#wakeTimer = undefined; + } + + await this.#autoscaledPool.abort(); + } + + /** + * Ends the run and marks it as history: the crawler stops reporting itself as running, and what is left is + * readable but no longer drivable. Ending here too means a subclass `teardown()` that forgets `super` cannot + * leave the task loop dispatching. + */ + async finish(): Promise { + await this.end(); + this.#finished = true; + } +} diff --git a/packages/browser-crawler/src/internals/browser-crawler.ts b/packages/browser-crawler/src/internals/browser-crawler.ts index aa531dda0dc1..7722871da46f 100644 --- a/packages/browser-crawler/src/internals/browser-crawler.ts +++ b/packages/browser-crawler/src/internals/browser-crawler.ts @@ -53,10 +53,14 @@ interface BaseResponse { /** * The type of a browser pool the crawler builds (and therefore owns) for itself. It's an {@apilink IBrowserPool} that - * additionally exposes `destroy()` — the crawler only ever tears down pools it created, which is why {@apilink IBrowserPool} - * itself intentionally omits `destroy`. + * additionally exposes the lifecycle hooks a crawler only ever calls on a pool it created — which is why + * {@apilink IBrowserPool} itself intentionally omits them: `releaseAllBrowsers()` at the end of every run, and + * `destroy()` once the crawler itself is destroyed. */ -export type OwnedBrowserPool = IBrowserPool & { destroy: () => Promise }; +export type OwnedBrowserPool = IBrowserPool & { + releaseAllBrowsers: () => Promise; + destroy: () => Promise; +}; /** * Rejects options that exist only to configure the browser pool the crawler would have built for itself. @@ -869,13 +873,18 @@ export abstract class BrowserCrawler< } /** - * Function for cleaning up after all requests are processed. - * @ignore + * Closes the browsers of a pool the crawler owns, so a finished run leaves none behind. The pool itself is + * crawler-lifetime and survives — destroying it here would hand a repeated `run()` a dead pool. */ override async teardown(): Promise { - await this.#browserPoolDep.ifOwned((pool) => pool.destroy()); + await this.#browserPoolDep.ifOwned((pool) => pool.releaseAllBrowsers()); await super.teardown(); } + + override async destroy(): Promise { + await super.destroy(); + await this.#browserPoolDep.ifOwned((pool) => pool.destroy()); + } } /** diff --git a/packages/browser-pool/src/browser-pool.ts b/packages/browser-pool/src/browser-pool.ts index b014b40fd5fa..b3bdb187a07f 100644 --- a/packages/browser-pool/src/browser-pool.ts +++ b/packages/browser-pool/src/browser-pool.ts @@ -773,7 +773,20 @@ export class BrowserPool< } /** - * Closes all managed browsers and tears down the pool. + * Closes every managed browser and empties the pool, which stays usable afterwards — a crawler releases its + * browsers when a run ends and may start another run on the same pool. + */ + async releaseAllBrowsers(): Promise { + await this.closeAllBrowsers(); + + this.startingBrowserControllers.clear(); + this.activeBrowserControllers.clear(); + this.retiredBrowserControllers.clear(); + } + + /** + * Closes all managed browsers and tears the pool down for good: its intervals are cleared without being + * re-armed and its listeners are dropped, so it cannot be used again. */ async destroy(): Promise { clearInterval(this.browserKillerInterval!); @@ -781,15 +794,7 @@ export class BrowserPool< this.browserKillerInterval = undefined; this.#browserRetireInterval = undefined; - await this.closeAllBrowsers(); - - this.teardown(); - } - - private teardown() { - this.startingBrowserControllers.clear(); - this.activeBrowserControllers.clear(); - this.retiredBrowserControllers.clear(); + await this.releaseAllBrowsers(); this.removeAllListeners(); } diff --git a/packages/browser-pool/src/remote-browser-pool.ts b/packages/browser-pool/src/remote-browser-pool.ts index e365766c4f22..00e81115b241 100644 --- a/packages/browser-pool/src/remote-browser-pool.ts +++ b/packages/browser-pool/src/remote-browser-pool.ts @@ -289,6 +289,14 @@ export class RemoteBrowserPool implements IBrowserPool { return this.#pool.injectPageState(page, state); } + /** + * Closes all browsers and releases their remote sessions, leaving the pool usable. Closing a browser is what + * releases its session (see the constructor), so this needs no separate release pass. + */ + async releaseAllBrowsers(): Promise { + await this.browserPool.releaseAllBrowsers(); + } + async [Symbol.asyncDispose](): Promise { await this.destroy(); } diff --git a/packages/core/src/log.ts b/packages/core/src/log.ts index e20763e2f8f1..26f5784abeb5 100644 --- a/packages/core/src/log.ts +++ b/packages/core/src/log.ts @@ -1,9 +1,9 @@ -import type { CrawleeLogger, CrawleeLoggerOptions } from '@crawlee/types'; +import type { CrawleeLogger, CrawleeLoggerOptions, LogOptions } from '@crawlee/types'; import type { LoggerOptions } from '@apify/log'; import log, { Log, Logger, LoggerJson, LoggerText, LogLevel } from '@apify/log'; -export type { CrawleeLogger, CrawleeLoggerOptions }; +export type { CrawleeLogger, CrawleeLoggerOptions, LogOptions }; /** * Abstract base class for custom Crawlee logger implementations. @@ -39,7 +39,7 @@ export abstract class BaseCrawleeLogger implements CrawleeLogger { // Note: If wrapping logger in a Proxy, unbound methods calling #-fields throw TypeError // unless bound to the target (see createLogProxy in adaptive-playwright-crawler.ts). #options: CrawleeLoggerOptions; - readonly #warningsLogged = new Set(); + readonly #loggedOnce = new Set(); constructor(options: Partial = {}) { this.#options = options; @@ -76,7 +76,30 @@ export abstract class BaseCrawleeLogger implements CrawleeLogger { return this.createChild(options); } - error(message: string, data?: Record): void { + /** + * Whether `once` suppresses this message. Keyed by level as well as text, so that a message logged once as + * `info` does not swallow the same text logged as a `warning`. + * + * Handled here rather than in {@apilink BaseCrawleeLogger.logWithLevel} so that every logger gets it, instead + * of each implementation having to dedupe for itself. + */ + #suppressedAsRepeat(level: number, message: string, options?: LogOptions): boolean { + if (!options?.once) { + return false; + } + + const key = `${level}:${message}`; + + if (this.#loggedOnce.has(key)) { + return true; + } + + this.#loggedOnce.add(key); + return false; + } + + error(message: string, data?: Record, options?: LogOptions): void { + if (this.#suppressedAsRepeat(LogLevel.ERROR, message, options)) return; this.logWithLevel(LogLevel.ERROR, message, data); } @@ -88,30 +111,34 @@ export abstract class BaseCrawleeLogger implements CrawleeLogger { }); } - softFail(message: string, data?: Record): void { + softFail(message: string, data?: Record, options?: LogOptions): void { + if (this.#suppressedAsRepeat(LogLevel.SOFT_FAIL, message, options)) return; this.logWithLevel(LogLevel.SOFT_FAIL, message, data); } - warning(message: string, data?: Record): void { + warning(message: string, data?: Record, options?: LogOptions): void { + if (this.#suppressedAsRepeat(LogLevel.WARNING, message, options)) return; this.logWithLevel(LogLevel.WARNING, message, data); } warningOnce(message: string): void { - if (!this.#warningsLogged.has(message)) { - this.#warningsLogged.add(message); - this.warning(message); - } + // Gates before delegating, so that a suppressed repeat is not even a `warning()` call. + if (this.#suppressedAsRepeat(LogLevel.WARNING, message, { once: true })) return; + this.warning(message); } - info(message: string, data?: Record): void { + info(message: string, data?: Record, options?: LogOptions): void { + if (this.#suppressedAsRepeat(LogLevel.INFO, message, options)) return; this.logWithLevel(LogLevel.INFO, message, data); } - debug(message: string, data?: Record): void { + debug(message: string, data?: Record, options?: LogOptions): void { + if (this.#suppressedAsRepeat(LogLevel.DEBUG, message, options)) return; this.logWithLevel(LogLevel.DEBUG, message, data); } - perf(message: string, data?: Record): void { + perf(message: string, data?: Record, options?: LogOptions): void { + if (this.#suppressedAsRepeat(LogLevel.PERF, message, options)) return; this.logWithLevel(LogLevel.PERF, `[PERF] ${message}`, data); } diff --git a/packages/playwright-crawler/src/internals/adaptive-playwright-crawler.ts b/packages/playwright-crawler/src/internals/adaptive-playwright-crawler.ts index 10f17c677fea..ff20d953b0b8 100644 --- a/packages/playwright-crawler/src/internals/adaptive-playwright-crawler.ts +++ b/packages/playwright-crawler/src/internals/adaptive-playwright-crawler.ts @@ -264,6 +264,12 @@ const proxyLogMethods = [ type LogProxyCall = [log: CrawleeLogger, method: (typeof proxyLogMethods)[number], ...args: unknown[]]; +/** The lifecycle {@apilink AdaptivePlaywrightCrawler} drives on the crawlers backing its context pipelines. */ +interface InnerCrawlerLifecycle { + teardown: () => Promise; + destroy: () => Promise; +} + /** * An extension of {@apilink PlaywrightCrawler} that uses a more limited request handler interface so that it is able to switch to HTTP-only crawling when it detects it may be possible. * @@ -322,7 +328,11 @@ export class AdaptivePlaywrightCrawler< */ readonly #attemptWritePolicy: Partial; - #teardownHooks: (() => Promise)[] = []; + /** Owns the browser pool this crawler's runs use, so its per-run resources are released with ours. */ + readonly #browserCrawler: InnerCrawlerLifecycle; + + /** Nothing of its state is per-run, but it owns a session pool that outlives one. */ + readonly #staticCrawler: InnerCrawlerLifecycle; constructor( options: AdaptivePlaywrightCrawlerOptions< @@ -470,7 +480,8 @@ export class AdaptivePlaywrightCrawler< remoteBrowser, }); - this.#teardownHooks.push(browserCrawler.teardown.bind(browserCrawler)); + this.#staticCrawler = staticCrawler; + this.#browserCrawler = browserCrawler; this.#staticContextPipeline = staticCrawler.contextPipeline.compose({ action: this.adaptCheerioContext.bind(this), @@ -844,9 +855,13 @@ export class AdaptivePlaywrightCrawler< // Mirrors the owned-only `initialize()` in `init()` - without this, the predictor we built keeps its // PERSIST_STATE listener registered after the crawl and never gets a final write. await this.#renderingTypePredictor.ifOwned((predictor) => predictor.teardown()); - for (const hook of this.#teardownHooks) { - await hook(); - } + await this.#browserCrawler.teardown(); + } + + override async destroy(): Promise { + await super.destroy(); + await this.#staticCrawler.destroy(); + await this.#browserCrawler.destroy(); } } diff --git a/packages/types/src/logger.ts b/packages/types/src/logger.ts index a3520c20f236..338e8ff12be9 100644 --- a/packages/types/src/logger.ts +++ b/packages/types/src/logger.ts @@ -6,6 +6,14 @@ export interface CrawleeLoggerOptions { prefix?: string | null; } +/** + * Options for a single log call. + */ +export interface LogOptions { + /** Log the message only the first time this logger sees that text, ignoring later calls with it. */ + once?: boolean; +} + /** * Interface for Crawlee logger implementations. * This allows users to inject custom loggers (e.g., Winston, Pino) while maintaining @@ -30,7 +38,7 @@ export interface CrawleeLogger { /** * Logs an `ERROR` message. */ - error(message: string, data?: Record): void; + error(message: string, data?: Record, options?: LogOptions): void; /** * Logs an `ERROR` level message with a nicely formatted exception. @@ -40,32 +48,32 @@ export interface CrawleeLogger { /** * Logs a `SOFT_FAIL` level message. */ - softFail(message: string, data?: Record): void; + softFail(message: string, data?: Record, options?: LogOptions): void; /** * Logs a `WARNING` level message. */ - warning(message: string, data?: Record): void; + warning(message: string, data?: Record, options?: LogOptions): void; /** - * Logs a `WARNING` level message only once. + * Logs a `WARNING` level message only once. Shorthand for `warning(message, undefined, { once: true })`. */ warningOnce(message: string): void; /** * Logs an `INFO` message. */ - info(message: string, data?: Record): void; + info(message: string, data?: Record, options?: LogOptions): void; /** * Logs a `DEBUG` message. */ - debug(message: string, data?: Record): void; + debug(message: string, data?: Record, options?: LogOptions): void; /** * Logs a `PERF` level message for performance tracking. */ - perf(message: string, data?: Record): void; + perf(message: string, data?: Record, options?: LogOptions): void; /** * Logs given message only once as WARNING for deprecated features. diff --git a/test/core/crawlers/basic_crawler.test.ts b/test/core/crawlers/basic_crawler.test.ts index bd12bf58e57e..28d99253401b 100644 --- a/test/core/crawlers/basic_crawler.test.ts +++ b/test/core/crawlers/basic_crawler.test.ts @@ -373,6 +373,41 @@ describe('BasicCrawler', () => { expect(secondSystem.desiredConcurrency).toBeLessThanOrEqual(2); }); + test('running tracks the run in progress', async () => { + let runningInHandler: boolean | undefined; + const crawler = new BasicCrawler({ + requestHandler: async () => { + runningInHandler = crawler.running; + }, + }); + + expect(crawler.running).toBe(false); + await crawler.run(['https://example.com/1']); + + expect(runningInHandler).toBe(true); + expect(crawler.running).toBe(false); + }); + + test('pause(), resume() and stop() warn once the run has finished', async () => { + const crawler = new BasicCrawler({ + requestHandler: async () => {}, + }); + + await crawler.run(['https://example.com/1']); + + const warning = vitest.spyOn(crawler.log, 'warning'); + await crawler.pause(); + crawler.resume(); + crawler.stop(); + + // The finished run's task loop is aborted, so driving it would do nothing while looking like it worked. + expect(warning.mock.calls.map(([message]) => message)).toEqual([ + 'Cannot pause a crawler that is not running.', + 'Cannot resume a crawler that is not running.', + 'Cannot stop a crawler that is not running.', + ]); + }); + test('stops the owned ConcurrencySystem when startup fails after it was started', async () => { const crawler = new BasicCrawler({ requestHandler: async () => {}, @@ -3241,7 +3276,7 @@ describe('BasicCrawler', () => { expect(enqueueLimitMessages).toHaveLength(2); }); - test('maxCrawlDepth limit log message should only be logged once per run', async () => { + test('maxCrawlDepth limit log message should only be logged once', async () => { const requestQueue = await RequestQueue.open(); // Each handler will try to add URLs that exceed maxCrawlDepth @@ -3264,15 +3299,16 @@ describe('BasicCrawler', () => { }, }); - const infoSpy = vitest.spyOn(crawler.log, 'info'); + // The `once` gate lives inside the logger, so `info()` is still called for every skipped request - + // what has to happen once is the message actually going out. + const logSpy = vitest.spyOn(crawler.log, 'logWithLevel'); // Run with two initial requests // Each will enqueue children at depth 1, then those children will try to enqueue at depth 2 (blocked) await crawler.run(['http://example.com/first', 'http://example.com/second']); - // The maxCrawlDepth limit message should only appear once per run, even though multiple requests triggered it - const maxCrawlDepthMessages = infoSpy.mock.calls.filter( - (call) => typeof call[0] === 'string' && call[0].includes('maxCrawlDepth'), + const maxCrawlDepthMessages = logSpy.mock.calls.filter( + (call) => typeof call[1] === 'string' && call[1].includes('maxCrawlDepth'), ); expect(maxCrawlDepthMessages).toHaveLength(1); }); diff --git a/test/core/crawlers/browser_crawler.test.ts b/test/core/crawlers/browser_crawler.test.ts index a61e6560d350..e3ad893c3b38 100644 --- a/test/core/crawlers/browser_crawler.test.ts +++ b/test/core/crawlers/browser_crawler.test.ts @@ -118,33 +118,63 @@ describe('BrowserCrawler', () => { }); }); - test.concurrent('should teardown browser pool', async () => { + test.concurrent('a run releases the browsers of an owned pool instead of destroying it', async () => { const puppeteerPlugin = new PuppeteerPlugin(puppeteer); - const requestList = await RequestList.open({ - sources: [{ url: 'http://example.com/?q=1' }], - }); const browserCrawler = new BrowserCrawlerTest({ browserPoolOptions: { browserPlugins: [puppeteerPlugin], }, - requestList, - + maxConcurrency: 1, requestHandler: async () => {}, maxRequestRetries: 1, }); - // Spy on destroy and track if it was called - let destroyCalled = false; const ownedPool = browserCrawler.browserPool as BrowserPool; - const originalDestroy = ownedPool.destroy.bind(ownedPool); - ownedPool.destroy = async () => { - destroyCalled = true; - return originalDestroy(); - }; + const releaseSpy = vitest.spyOn(ownedPool, 'releaseAllBrowsers'); + const destroySpy = vitest.spyOn(ownedPool, 'destroy'); + ownedPool.on(BROWSER_POOL_EVENTS.BROWSER_LAUNCHED, () => {}); + + await browserCrawler.run([`${serverAddress}/?q=1`]); + + expect(releaseSpy).toHaveBeenCalled(); + expect(destroySpy).not.toHaveBeenCalled(); + // What a destroyed pool loses for good, since nothing re-arms either: its listeners, and the timers that + // retire idle browsers and reap the retired ones. + expect(ownedPool.listenerCount(BROWSER_POOL_EVENTS.BROWSER_LAUNCHED)).toBe(1); + // eslint-disable-next-line dot-notation -- TS-private on the pool + expect(ownedPool['browserKillerInterval']).toBeDefined(); + + await browserCrawler.destroy(); + expect(destroySpy).toHaveBeenCalledTimes(1); + }); - await browserCrawler.run(); - expect(destroyCalled).toBe(true); + test.concurrent('a repeated run() crawls with the same browser pool', async () => { + const puppeteerPlugin = new PuppeteerPlugin(puppeteer); + + const processed: string[] = []; + const browserCrawler = new BrowserCrawlerTest({ + browserPoolOptions: { + browserPlugins: [puppeteerPlugin], + }, + maxConcurrency: 1, + requestHandler: async ({ request }) => { + processed.push(request.url); + }, + }); + + const launched: unknown[] = []; + (browserCrawler.browserPool as BrowserPool).on(BROWSER_POOL_EVENTS.BROWSER_LAUNCHED, (controller) => { + launched.push(controller); + }); + + await browserCrawler.run([`${serverAddress}/?q=1`]); + await browserCrawler.run([`${serverAddress}/?q=2`]); + + expect(processed).toEqual([`${serverAddress}/?q=1`, `${serverAddress}/?q=2`]); + // Each run launches its own browser, because the previous one released its browsers on the way out - and + // the pool is still the crawler's, so it still reports the launch. + expect(launched).toHaveLength(2); }); test.concurrent('should not tear down a user-supplied browser pool', async () => { From d78e358a3b2c9cfec5bbbdbe8e62305fa743c51d Mon Sep 17 00:00:00 2001 From: Jan Buchar Date: Tue, 8 Sep 2026 14:49:50 +0200 Subject: [PATCH 2/3] Address review feedback --- docs/upgrading/upgrading_v4.md | 8 ++++--- .../src/internals/basic-crawler.ts | 22 +++++++++---------- test/core/crawlers/basic_crawler.test.ts | 22 +++++++++++++++++++ 3 files changed, 38 insertions(+), 14 deletions(-) diff --git a/docs/upgrading/upgrading_v4.md b/docs/upgrading/upgrading_v4.md index 6243a05664fc..d8d6dab94e56 100644 --- a/docs/upgrading/upgrading_v4.md +++ b/docs/upgrading/upgrading_v4.md @@ -659,10 +659,12 @@ An alias identifies a run-scoped queue. It has no persistent name, and is emptie What outlives a run is released by `crawler.destroy()`, or by disposing of the crawler: ```typescript -await using crawler = new PlaywrightCrawler({ requestHandler: async ({ page }) => { /* ... */ } }); +{ + await using crawler = new PlaywrightCrawler({ requestHandler: async ({ page }) => { /* ... */ } }); -await crawler.run(['https://example.com/a']); -await crawler.run(['https://example.com/b']); + await crawler.run(['https://example.com/a']); + await crawler.run(['https://example.com/b']); +} // the browser pool is destroyed here, as the crawler goes out of scope ``` Disposing is optional — a finished run leaves no browsers open and no timer holding the process alive. diff --git a/packages/basic-crawler/src/internals/basic-crawler.ts b/packages/basic-crawler/src/internals/basic-crawler.ts index 42c9043c4a1f..f12034101099 100644 --- a/packages/basic-crawler/src/internals/basic-crawler.ts +++ b/packages/basic-crawler/src/internals/basic-crawler.ts @@ -1723,19 +1723,19 @@ export class BasicCrawler< this.#run = run; - await purgeDefaultStorages({ - onlyPurgeOnce: true, - storageBackend: serviceLocator.getStorageBackend(), - configuration: serviceLocator.getConfiguration(), - }); + try { + await purgeDefaultStorages({ + onlyPurgeOnce: true, + storageBackend: serviceLocator.getStorageBackend(), + configuration: serviceLocator.getConfiguration(), + }); - if (requests) { - await this.addRequests(requests, options); - } + if (requests) { + await this.addRequests(requests, options); + } - try { - // Inside the try: from here on, a failed startup has something to stop again. An injected - // governor is the caller's to run, and the task loop rejects an unstarted one. + // An injected governor is the caller's to run; an owned one is started here and stopped again + // below if the startup fails. Either way the task loop rejects one that is not running. await this.#concurrencySystem.ifOwned((system) => system.start()); await this.init(); await this.statistics.startCapturing(); diff --git a/test/core/crawlers/basic_crawler.test.ts b/test/core/crawlers/basic_crawler.test.ts index 28d99253401b..08bbb5de34d7 100644 --- a/test/core/crawlers/basic_crawler.test.ts +++ b/test/core/crawlers/basic_crawler.test.ts @@ -431,6 +431,28 @@ describe('BasicCrawler', () => { await crawler.run(['https://example.com/2']); }); + test('a startup that fails before the crawl leaves the instance runnable', async () => { + const processed: string[] = []; + const crawler = new BasicCrawler({ + requestHandler: async ({ request }) => { + processed.push(request.url); + }, + }); + + const failure = new Error('Could not add the initial requests'); + // Enqueueing the initial requests happens before the crawl starts, and used to happen outside the + // startup's failure handling - leaving the instance wedged as `running` for good. + const addRequests = vitest.spyOn(crawler, 'addRequests').mockRejectedValue(failure); + + await expect(crawler.run(['https://example.com/1'])).rejects.toThrow(failure); + expect(crawler.running).toBe(false); + + addRequests.mockRestore(); + await crawler.run(['https://example.com/2']); + + expect(processed).toEqual(['https://example.com/2']); + }); + test('should process 4 requests total when calling run() twice with maxRequestsPerCrawl: 2', async () => { const processed: { url: string }[] = []; From 2261afe8d90c66aee445406e702a80c138c9b00e Mon Sep 17 00:00:00 2001 From: Jan Buchar Date: Tue, 8 Sep 2026 15:50:16 +0200 Subject: [PATCH 3/3] Update API snapshot --- docs/public-api/crawlee-playwright.api.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/public-api/crawlee-playwright.api.md b/docs/public-api/crawlee-playwright.api.md index 22fc52f63eed..3d9a7a9d541a 100644 --- a/docs/public-api/crawlee-playwright.api.md +++ b/docs/public-api/crawlee-playwright.api.md @@ -74,13 +74,13 @@ export class AdaptivePlaywrightCrawler, Ext constructor(options?: AdaptivePlaywrightCrawlerOptions); // (undocumented) protected buildContextPipeline(): ContextPipeline_2; + // (undocumented) + destroy(): Promise; drainRenderingDetections(input?: { timeoutMillis?: number; }): Promise; get inFlightRenderingTypeDetectionCount(): number; // (undocumented) - destroy(): Promise; - // (undocumented) protected init(): Promise; // (undocumented) protected runRequestHandler(crawlingContext: CrawlingContext_2): Promise;