diff --git a/packages/basic-crawler/src/internals/basic-crawler.ts b/packages/basic-crawler/src/internals/basic-crawler.ts index 2dbc47698f24..6663e784d29d 100644 --- a/packages/basic-crawler/src/internals/basic-crawler.ts +++ b/packages/basic-crawler/src/internals/basic-crawler.ts @@ -575,7 +575,7 @@ export class BasicCrawler(); private experiments: CrawlerExperiments; private readonly robotsTxtFileCache: LruCache; @@ -1042,8 +1042,8 @@ export class BasicCrawler { if (!this.events.isInitialized()) { await this.events.init(); - this._closeEvents = true; + this._ownsEventManager = true; } this.autoscaledPool = new AutoscaledPool(this.autoscaledPoolOptions, this.config); @@ -2031,13 +2031,18 @@ export class BasicCrawler { - this.events.emit(EventType.PERSIST_STATE, { isMigrating: false }); + // 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._ownsEventManager) { + this.events.emit(EventType.PERSIST_STATE, { isMigrating: false }); + } if (this.useSessionPool) { - await this.sessionPool!.teardown(); + await this.sessionPool!.teardown({ persistState: this._ownsEventManager }); } - if (this._closeEvents) { + if (this._ownsEventManager) { await this.events.close(); } diff --git a/packages/core/src/session_pool/session_pool.ts b/packages/core/src/session_pool/session_pool.ts index 22528a29ebbd..1a57ed3c75ba 100644 --- a/packages/core/src/session_pool/session_pool.ts +++ b/packages/core/src/session_pool/session_pool.ts @@ -384,10 +384,13 @@ export class SessionPool extends EventEmitter { /** * Removes listener from `persistState` event. * This function should be called after you are done with using the `SessionPool` instance. + * @param options - Set `persistState` to false when the final state was already persisted by the event manager. */ - async teardown(): Promise { + async teardown({ persistState = true }: { persistState?: boolean } = {}): Promise { this.events.off(EventType.PERSIST_STATE, this._listener); - await this.persistState(); + if (persistState) { + await this.persistState(); + } } /** diff --git a/test/core/crawlers/basic_crawler.test.ts b/test/core/crawlers/basic_crawler.test.ts index 72c572560532..9844470e231d 100644 --- a/test/core/crawlers/basic_crawler.test.ts +++ b/test/core/crawlers/basic_crawler.test.ts @@ -26,7 +26,7 @@ import { RequestValidationError, Router, } from '@crawlee/basic'; -import { RequestState } from '@crawlee/core'; +import { RequestState, SessionPool, Statistics } from '@crawlee/core'; import type { Dictionary } from '@crawlee/utils'; import { RobotsTxtFile, sleep } from '@crawlee/utils'; import express from 'express'; @@ -1337,6 +1337,52 @@ describe('BasicCrawler', () => { }); describe('Uses SessionPool', () => { + it('persists statistics and the session pool once when finishing', async () => { + const persistStatistics = vitest.spyOn(Statistics.prototype, 'persistState'); + const persistSessionPool = vitest.spyOn(SessionPool.prototype, 'persistState'); + const config = new Configuration(); + const crawler = new BasicCrawler( + { + useSessionPool: true, + requestHandler: async () => { + persistStatistics.mockClear(); + persistSessionPool.mockClear(); + }, + }, + config, + ); + + await crawler.run(['https://example.com']); + + expect(persistStatistics).toHaveBeenCalledTimes(1); + expect(persistSessionPool).toHaveBeenCalledTimes(1); + }); + + it('persists the session pool once with a shared event manager', async () => { + const persistStatistics = vitest.spyOn(Statistics.prototype, 'persistState'); + const persistSessionPool = vitest.spyOn(SessionPool.prototype, 'persistState'); + const config = new Configuration(); + await config.getEventManager().init(); + const crawler = new BasicCrawler( + { + useSessionPool: true, + requestHandler: async () => { + persistStatistics.mockClear(); + persistSessionPool.mockClear(); + }, + }, + config, + ); + + await crawler.run(['https://example.com']); + expect(config.getEventManager().listenerCount(EventType.PERSIST_STATE)).toBe(0); + expect(persistSessionPool).toHaveBeenCalledTimes(1); + await config.getEventManager().close(); + + expect(persistStatistics).toHaveBeenCalledTimes(1); + expect(persistSessionPool).toHaveBeenCalledTimes(1); + }); + it('should use SessionPool when useSessionPool is true ', async () => { const url = 'https://example.com'; const requestList = await RequestList.open({ sources: [{ url }] });