[pull] master from apify:master

docs/public-api/crawlee-core.api.md

@@ -417,7 +417,6 @@ export interface IRequestLoader {
     getPendingCount(): Promise<number>;
     getTotalCount(): Promise<number>;
     markRequestAsHandled(request: Request_2): Promise<RequestQueueOperationInfo | void | null>;
-    persistState?(): Promise<void>;
     toTandem?(requestManager?: IRequestManager): Promise<IRequestManager>;
 }
 
@@ -816,7 +815,6 @@ export class RequestManagerTandem implements IRequestManager {
     getTotalCount(): Promise<number>;
     // (undocumented)
     markRequestAsHandled(request: Request_2): Promise<RequestQueueOperationInfo | void | null>;
-    persistState(): Promise<void>;
     purge(): Promise<void>;
     // (undocumented)
     reclaimRequest(request: Request_2, options?: RequestQueueOperationOptions): Promise<RequestQueueOperationInfo | null>;

docs/upgrading/upgrading_v4.md

@@ -1466,7 +1466,7 @@ The harmonized loader interface differs from the old `IRequestList` in a few way
 | `isEmpty(): Promise<boolean>` and `isFinished(): Promise<boolean>` | `checkReadiness(): Promise<RequestSourceStatus>` ([details](#isempty--isfinished-replaced-by-checkreadiness)) |
 | `reclaimRequest()` on the interface | Removed from the read-only loaders entirely; reclaiming is a write operation that lives only on `IRequestManager` (e.g. `RequestQueue`, `RequestManagerTandem`) |
 | `inProgress: Set<string>` on the interface | Removed from the interface |
-| `persistState(): Promise<void>` (required) | `persistState?(): Promise<void>` (optional) |
+| `persistState(): Promise<void>` (required) | Removed from the interface; loaders that have state persist it themselves on the `persistState` event, and `RequestList`/`SitemapRequestLoader` still expose the method as a class member |
 | _(n/a)_ | `toTandem?(requestManager?)` (new) |
 
 `RequestList.length()` and `RequestList.handledCount()` (and their `SitemapRequestLoader` counterparts) were renamed to `getTotalCount()` and `getHandledCount()` and are now `async` — `await` them.

packages/basic-crawler/src/internals/basic-crawler.ts

@@ -1107,8 +1107,7 @@ export class BasicCrawler<
                 : suppliedManager;
 
             if (requestList !== undefined) {
-                // The list is read first, while new requests still have somewhere writable to go; the tandem also
-                // forwards `persistState()` to the loader.
+                // The list is read first, while new requests still have somewhere writable to go.
                 this.requestManager = new RequestManagerTandem(
                     requestList,
                     writableManager ?? (() => this.openOwnedRequestQueue()),
@@ -2606,30 +2605,9 @@ export class BasicCrawler<
             });
         }
 
-        const requestManagerPersistPromise = (async () => {
-            // The request manager persists its read-only loader's state, if it has one that supports
-            // persistence (e.g. a tandem wrapping a `RequestList`). For a plain `RequestQueue`, this is a no-op.
-            if (this.requestManager?.persistState) {
-                if ((await this.requestManager.checkReadiness()).status === 'finished') return;
-                await this.requestManager.persistState().catch((err) => {
-                    if (err.message.includes('Cannot persist state.')) {
-                        this.log.error(
-                            "The crawler attempted to persist its request list's state and failed due to missing or " +
-                                'invalid configuration. Make sure to use either RequestList.open() or the "stateKeyPrefix" option of RequestList ' +
-                                'constructor to ensure your crawling state is persisted through host migrations and restarts.',
-                        );
-                    } else {
-                        this.log.exception(
-                            err,
-                            'An unexpected error occurred when the crawler ' +
-                                "attempted to persist its request list's state.",
-                        );
-                    }
-                });
-            }
-        })();
-
-        await Promise.all([requestManagerPersistPromise, this.statistics.persistState?.()]);
+        // Captures the statistics changed while draining above. No PERSIST_STATE event is tied to the drain:
+        // the periodic one is unrelated, and the one the platform emits on migration arrives before any of this.
+        await this.statistics.persistState?.();
     }
 
     /**

packages/basic-crawler/src/internals/throttling_request_manager.ts

@@ -1180,10 +1180,6 @@ export class ThrottlingRequestManager<T extends IRequestManager = IRequestManage
         }
     }
 
-    async persistState(): Promise<void> {
-        await this.#forEachManager((manager) => manager.persistState?.());
-    }
-
     async drop(): Promise<void> {
         await this.#forEachManager((manager) => (manager as { drop?(): Promise<void> }).drop?.());
         this.#subManagers.clear();

packages/basic-crawler/test/throttling_request_manager.test.ts

@@ -383,7 +383,6 @@ describe('ThrottlingRequestManager', () => {
             const manager = new ThrottlingRequestManager({ ...throttling, inner: factory });
 
             await manager.purge();
-            await manager.persistState();
             await manager.setExpectedRequestProcessingTimeSecs(600);
             await manager.drop();
 

packages/core/src/recoverable_state.ts

@@ -1,6 +1,10 @@
 import { addTimeoutToPromise, storage as timeoutStorage } from '@apify/timeout';
-import type { Configuration, CrawleeLogger } from '@crawlee/core';
-import { EventType, KeyValueStore, serviceLocator, StateValidationError } from '@crawlee/core';
+import type { Configuration } from './configuration.js';
+import { StateValidationError } from './errors.js';
+import { EventType } from './events/event_manager.js';
+import type { CrawleeLogger } from './log.js';
+import { serviceLocator } from './service_locator.js';
+import { KeyValueStore } from './storages/key_value_store.js';
 import type { Awaitable } from '@crawlee/types';
 import type { StandardSchemaV1 } from '@standard-schema/spec';
 

packages/core/src/storages/request_loader.ts

@@ -89,7 +89,7 @@ export function joinRequestSourceStatuses(a: RequestSourceStatus, b: RequestSour
  * "finished with this request", whether processing succeeded or was abandoned after exhausting retries.
  *
  * Honoring this contract matters for three reasons:
- * - **Restarts and migrations:** loaders that persist their state (see {@apilink IRequestLoader.persistState})
+ * - **Restarts and migrations:** loaders that persist their state (such as {@apilink RequestList})
  *   treat in-progress requests as interrupted and re-serve them after a restart. A request that is fetched
  *   but never marked handled will be crawled again.
  * - **Termination detection:** {@apilink IRequestLoader.checkReadiness} only reports `finished` once nothing is
@@ -154,13 +154,6 @@ export interface IRequestLoader {
      */
     markRequestAsHandled(request: Request): Promise<RequestQueueOperationInfo | void | null>;
 
-    /**
-     * Persists the current state of the loader into the default {@apilink KeyValueStore}.
-     *
-     * Not all loaders support persistence; implementations that do not should leave this `undefined`.
-     */
-    persistState?(): Promise<void>;
-
     /**
      * Combines the loader with a request manager to support adding and reclaiming requests.
      *

packages/core/src/storages/request_manager_tandem.ts

@@ -217,14 +217,6 @@ export class RequestManagerTandem implements IRequestManager {
         return (await this.getRequestManager()).addRequestsBatched(requests, options);
     }
 
-    /**
-     * Persists the state of the underlying read-only loader, if it supports persistence.
-     * @inheritdoc
-     */
-    async persistState(): Promise<void> {
-        await this.#requestLoader.persistState?.();
-    }
-
     /**
      * Purges the writable request manager so the tandem can be reused (e.g. across repeated `crawler.run()` calls).
      * The read-only loader is immutable and cannot be purged, so only the manager side is reset.

packages/core/src/validators.ts

@@ -16,7 +16,7 @@ export const validators = {
         "Expected an object implementing the IProxyConfiguration interface (missing 'newProxyInfo'), got something else.",
     ),
     requestList: schemas.objectWithKeys(
-        ['fetchNextRequest', 'persistState'],
+        ['fetchNextRequest', 'checkReadiness'],
         'Expected a RequestList, got something else.',
     ),
     requestQueue: schemas.objectWithKeys(

test/core/crawlers/basic_crawler.test.ts

@@ -1081,7 +1081,10 @@ describe('BasicCrawler', () => {
             const processed: { url: string }[] = [];
             const requestList = await RequestList.open('reqList', sources);
             const requestHandler: RequestHandler = async ({ request }) => {
-                if (request.url.endsWith('200')) serviceLocator.getEventManager().emit(event);
+                if (request.url.endsWith('200')) {
+                    serviceLocator.getEventManager().emit(event);
+                    serviceLocator.getEventManager().emit(EventType.PERSIST_STATE);
+                }
                 processed.push({ url: request.url });
             };
 

test/core/request_manager_tandem.test.ts

@@ -331,18 +331,6 @@ describe('RequestManagerTandem', () => {
         ).toBe(false);
     });
 
-    test('persistState forwards to the read-only loader', async () => {
-        const requestList = await RequestList.open(null, [{ url: 'https://example.com/1' }]);
-        const requestQueue = await RequestQueue.open();
-
-        const persistSpy = vi.spyOn(requestList, 'persistState').mockResolvedValue();
-
-        const tandem = new RequestManagerTandem(requestList, requestQueue);
-        await tandem.persistState();
-
-        expect(persistSpy).toHaveBeenCalledTimes(1);
-    });
-
     test('setExpectedRequestProcessingTimeSecs forwards to an already-resolved manager', async () => {
         const requestList = await RequestList.open(null, [{ url: 'https://example.com/1' }]);
         const requestQueue = await RequestQueue.open();