diff --git a/packages/basic-crawler/src/internals/basic-crawler.ts b/packages/basic-crawler/src/internals/basic-crawler.ts index 0ec4962a643b..243192914754 100644 --- a/packages/basic-crawler/src/internals/basic-crawler.ts +++ b/packages/basic-crawler/src/internals/basic-crawler.ts @@ -845,6 +845,7 @@ export class BasicCrawler< running = false; hasFinishedBefore = false; #unexpectedStop = false; + #teardownRequested = false; #log!: CrawleeLogger; @@ -1707,6 +1708,7 @@ export class BasicCrawler< 'This crawler instance is already running, you can add more requests to it via `crawler.addRequests()`.', ); } + this.#teardownRequested = false; const { purgeRequestQueue, ...addRequestsOptions } = options ?? {}; @@ -1795,7 +1797,7 @@ export class BasicCrawler< let stats = {} as FinalStatistics; try { - await this.#autoscaledPool!.run(); + if (!this.#teardownRequested) await this.#autoscaledPool!.run(); } finally { await this.statistics.stopCapturing(); await this.teardown(); @@ -2996,6 +2998,8 @@ export class BasicCrawler< * To stop the crawler gracefully (waiting for all running requests to finish), use {@apilink BasicCrawler.stop|`crawler.stop()`} instead. */ async teardown(): Promise { + this.#teardownRequested = true; + // 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. diff --git a/test/core/crawlers/basic_crawler.test.ts b/test/core/crawlers/basic_crawler.test.ts index 0a785a28cff6..62fac7f96355 100644 --- a/test/core/crawlers/basic_crawler.test.ts +++ b/test/core/crawlers/basic_crawler.test.ts @@ -36,7 +36,7 @@ import { ThrottlingRequestManager, } from '@crawlee/basic'; import type { CalculatedStatistics, IConcurrencySystem, IStatistics } from '@crawlee/core'; -import { ConcurrencySystem, MemoryStorageBackend, RequestState } from '@crawlee/core'; +import { ConcurrencySystem, LocalEventManager, MemoryStorageBackend, RequestState } from '@crawlee/core'; import { BaseHttpClient } from '@crawlee/http-client'; import type { Dictionary, ISession, ProxyInfo } from '@crawlee/types'; import { RobotsTxtFile, sleep } from '@crawlee/utils'; @@ -390,6 +390,101 @@ describe('BasicCrawler', () => { await crawler.run(['https://example.com/2']); }); + describe('teardown during startup', () => { + test.each([ + ['crawler-owned', false], + ['injected', true], + ])('teardown during startup settles run() with a %s ConcurrencySystem', async (_label, injectSystem) => { + const initialized = Promise.withResolvers(); + const resumeInitialization = Promise.withResolvers(); + const requestManager = await RequestQueue.open(`delayed-startup-${injectSystem}`); + vitest.spyOn(requestManager, 'setExpectedRequestProcessingTimeSecs').mockImplementation(async () => { + initialized.resolve(); + await resumeInitialization.promise; + }); + + const concurrencySystem = injectSystem ? new ConcurrencySystem({ maxConcurrency: 1 }) : undefined; + await concurrencySystem?.start(); + + const crawler = new BasicCrawler({ + keepAlive: true, + ...(concurrencySystem && { concurrencySystem }), + requestManager, + requestHandler: async () => {}, + }); + const runPromise = crawler.run().then( + () => 'resolved', + (error: Error) => `rejected: ${error.message}`, + ); + + await initialized.promise; + await crawler.teardown(); + resumeInitialization.resolve(); + + const outcome = await Promise.race([runPromise, sleep(500).then(() => 'still pending')]); + + // Clean up the broken behavior so the failing test does not leak the pool interval. + if (outcome === 'still pending') await crawler.teardown(); + await concurrencySystem?.stop(); + + expect(outcome).toBe('resolved'); + }); + + test.each([ + ['crawler-owned', false], + ['injected', true], + ])('teardown before pool creation settles run() with a %s ConcurrencySystem', async (_label, injectSystem) => { + const initializationStarted = Promise.withResolvers(); + const resumeInitialization = Promise.withResolvers(); + const eventManager = LocalEventManager.fromConfiguration(); + const initializeEventManager = eventManager.init.bind(eventManager); + vitest.spyOn(eventManager, 'init').mockImplementation(async () => { + initializationStarted.resolve(); + await resumeInitialization.promise; + await initializeEventManager(); + }); + + const concurrencySystem = injectSystem ? new ConcurrencySystem({ maxConcurrency: 1 }) : undefined; + await concurrencySystem?.start(); + + const crawler = new BasicCrawler({ + keepAlive: true, + ...(concurrencySystem && { concurrencySystem }), + eventManager, + requestHandler: async () => {}, + }); + const runPromise = crawler.run().then( + () => 'resolved', + (error: Error) => `rejected: ${error.message}`, + ); + + await initializationStarted.promise; + await crawler.teardown(); + resumeInitialization.resolve(); + const outcome = await Promise.race([runPromise, sleep(500).then(() => 'still pending')]); + + // Clean up the broken behavior so the failing test does not leak the pool interval. + if (outcome === 'still pending') await crawler.teardown(); + await concurrencySystem?.stop(); + + expect(outcome).toBe('resolved'); + }); + }); + + test('teardown before run() does not cancel the next run', async () => { + const processed: string[] = []; + const crawler = new BasicCrawler({ + requestHandler: async ({ request }) => { + processed.push(request.url); + }, + }); + + await crawler.teardown(); + await crawler.run(['https://example.com/after-teardown']); + + expect(processed).toEqual(['https://example.com/after-teardown']); + }); + test('should process 4 requests total when calling run() twice with maxRequestsPerCrawl: 2', async () => { const processed: { url: string }[] = [];