feat: Request Queue v2

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

@@ -15,6 +15,7 @@ import type {
     Request,
     RequestList,
     RequestOptions,
+    RequestProvider,
     RouterHandler,
     RouterRoutes,
     Session,
@@ -23,24 +24,25 @@ import type {
     StatisticState,
 } from '@crawlee/core';
 import {
-    mergeCookies,
+    Dataset,
     AutoscaledPool,
     Configuration,
-    enqueueLinks,
     EventType,
     KeyValueStore,
-    CriticalError,
     NonRetryableError,
     RequestQueue,
+    RequestQueueV2,
     RequestState,
+    RetryRequestError,
     Router,
     SessionPool,
     Statistics,
+    enqueueLinks,
+    mergeCookies,
     purgeDefaultStorages,
     validators,
-    RetryRequestError,
     SessionError,
-    Dataset,
+    CriticalError,
 } from '@crawlee/core';
 import type { Dictionary, Awaitable, BatchAddRequestsResult, SetStatusMessageOptions } from '@crawlee/types';
 import { ROTATE_PROXY_ERRORS } from '@crawlee/utils';
@@ -165,7 +167,7 @@ export interface BasicCrawlerOptions<Context extends CrawlingContext = BasicCraw
      * > Alternatively, `requests` parameter of {@apilink BasicCrawler.run|`crawler.run()`} could be used to enqueue the initial requests -
      * it is a shortcut for running `crawler.addRequests()` before the `crawler.run()`.
      */
-    requestQueue?: RequestQueue;
+    requestQueue?: RequestProvider;
 
     /**
      * Timeout in which the function passed as {@apilink BasicCrawlerOptions.requestHandler|`requestHandler`} needs to finish, in seconds.
@@ -325,6 +327,26 @@ export interface BasicCrawlerOptions<Context extends CrawlingContext = BasicCraw
 
     /** @internal */
     log?: Log;
+
+    /**
+     * Enables experimental features of Crawlee, which can alter the behavior of the crawler.
+     * WARNING: these options are not guaranteed to be stable and may change or be removed at any time.
+     */
+    experiments?: CrawlerExperiments;
+}
+
+/**
+ * A set of options that you can toggle to enable experimental features in Crawlee.
+ *
+ * NOTE: These options will not respect semantic versioning and may be removed or changed at any time. Use at your own risk.
+ * If you do use these and encounter issues, please report them to us.
+ */
+export interface CrawlerExperiments {
+    /**
+     * Enables the use of the new RequestQueue API, which allows multiple clients to use the same queue,
+     * by locking the requests they are processing for a period of time.
+     */
+    requestLocking?: boolean;
 }
 
 /**
@@ -410,7 +432,7 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
      * A reference to the underlying {@apilink RequestQueue} class that manages the crawler's {@apilink Request|requests}.
      * Only available if used by the crawler.
      */
-    requestQueue?: RequestQueue;
+    requestQueue?: RequestProvider;
 
     /**
      * A reference to the underlying {@apilink SessionPool} class that manages the crawler's {@apilink Session|sessions}.
@@ -456,6 +478,9 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
     protected retryOnBlocked: boolean;
     private _closeEvents?: boolean;
 
+    private experiments: CrawlerExperiments;
+    private _experimentWarnings: Partial<Record<keyof CrawlerExperiments, boolean>> = {};
+
     protected static optionsShape = {
         requestList: ow.optional.object.validate(validators.requestList),
         requestQueue: ow.optional.object.validate(validators.requestQueue),
@@ -493,6 +518,7 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
 
         // internal
         log: ow.optional.object,
+        experiments: ow.optional.object,
     };
 
     /**
@@ -522,6 +548,7 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
 
             // internal
             log = defaultLog.child({ prefix: this.constructor.name }),
+            experiments = {},
 
             // Old and new request handler methods
             handleRequestFunction,
@@ -546,6 +573,15 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
         this.statusMessageCallback = statusMessageCallback as StatusMessageCallback;
         this.events = config.getEventManager();
         this.domainAccessedTime = new Map();
+        this.experiments = experiments;
+
+        if (requestQueue && requestQueue instanceof RequestQueueV2 && !experiments.requestLocking) {
+            throw new Error([
+                'You provided the new RequestQueue v2 class into your crawler without enabling the experiment!',
+                "If you're sure you want to test out the new experimental RequestQueue v2, please provide `experiments: { requestLocking: true }` "
+                + 'in your crawler options, and try again.',
+            ].join('\n'));
+        }
 
         this._handlePropertyNameChange({
             newName: 'requestHandler',
@@ -596,9 +632,13 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
         const tryEnv = (val?: string) => (val == null ? null : +val);
         // allow at least 5min for internal timeouts
         this.internalTimeoutMillis = tryEnv(process.env.CRAWLEE_INTERNAL_TIMEOUT) ?? Math.max(this.requestHandlerTimeoutMillis * 2, 300e3);
+
         // override the default internal timeout of request queue to respect `requestHandlerTimeoutMillis`
         if (this.requestQueue) {
             this.requestQueue.internalTimeoutMillis = this.internalTimeoutMillis;
+            // for request queue v2, we want to lock requests by the timeout that would also account for internals (plus 5 seconds padding), but
+            // with a minimum of a minute
+            this.requestQueue.requestLockSecs = Math.max(this.internalTimeoutMillis / 1000 + 5, 60);
         }
 
         this.maxRequestRetries = maxRequestRetries;
@@ -769,7 +809,7 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
         // ignored - as a failed requests is still handled.
         if (this.running === false && this.requestQueue?.name === 'default' && purgeRequestQueue) {
             await this.requestQueue.drop();
-            this.requestQueue = await RequestQueue.open(null, { config: this.config });
+            this.requestQueue = await this._getRequestQueue();
         }
 
         this.running = true;
@@ -851,7 +891,7 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
     }
 
     async getRequestQueue() {
-        this.requestQueue ??= await RequestQueue.open(null, { config: this.config });
+        this.requestQueue ??= await this._getRequestQueue();
 
         return this.requestQueue!;
     }
@@ -1006,7 +1046,7 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
      * adding it back to the queue after the timeout passes. Returns `true` if the request
      * should be ignored and will be reclaimed to the queue once ready.
      */
-    protected delayRequest(request: Request, source: RequestQueue | RequestList) {
+    protected delayRequest(request: Request, source: RequestList | RequestProvider) {
         const domain = getDomain(request.url);
 
         if (!domain || !request) {
@@ -1021,13 +1061,15 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
             return false;
         }
 
-        source.inProgress.delete(request.id!);
+        // eslint-disable-next-line dot-notation
+        source['inProgress'].delete(request.id!);
         const delay = lastAccessTime + this.sameDomainDelayMillis - now;
         this.log.debug(`Request ${request.url} (${request.id}) will be reclaimed after ${delay} milliseconds due to same domain delay`);
         setTimeout(async () => {
             this.log.debug(`Adding request ${request.url} (${request.id}) back to the queue`);
-            source?.inProgress.add(request.id!);
-            await source?.reclaimRequest(request);
+            // eslint-disable-next-line dot-notation
+            source['inProgress'].add(request.id!);
+            await source.reclaimRequest(request);
         }, delay);
 
         return true;
@@ -1230,7 +1272,7 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
     protected async _requestFunctionErrorHandler(
         error: Error,
         crawlingContext: Context,
-        source: RequestList | RequestQueue,
+        source: RequestList | RequestProvider,
     ): Promise<void> {
         const { request } = crawlingContext;
         request.pushErrorMessage(error);
@@ -1444,6 +1486,22 @@ export class BasicCrawler<Context extends CrawlingContext = BasicCrawlingContext
 
         return request.headers?.Cookie || request.headers?.cookie || '';
     }
+
+    private _getRequestQueue() {
+        if (this.experiments.requestLocking) {
+            if (!this._experimentWarnings.requestLocking) {
+                this.log.warning([
+                    'The RequestQueue v2 is an experimental feature, and may have issues when used in a production environment.',
+                    'Please report any issues you encounter on GitHub: https://github.com/apify/crawlee',
+                ].join('\n'));
+                this._experimentWarnings.requestLocking = true;
+            }
+
+            return RequestQueueV2.open(null, { config: this.config });
+        }
+
+        return RequestQueue.open(null, { config: this.config });
+    }
 }
 
 export interface CreateContextOptions {

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

@@ -9,7 +9,7 @@ import type {
     ProxyConfiguration,
     ProxyInfo,
     RequestHandler,
-    RequestQueue,
+    RequestProvider,
     Session,
 } from '@crawlee/basic';
 import {
@@ -712,7 +712,7 @@ export abstract class BrowserCrawler<
 interface EnqueueLinksInternalOptions {
     options?: EnqueueLinksOptions;
     page: CommonPage;
-    requestQueue: RequestQueue;
+    requestQueue: RequestProvider;
     originalRequestUrl: string;
     finalRequestUrl?: string;
 }

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

@@ -1,18 +1,24 @@
 import type { IncomingMessage } from 'http';
 
 import type {
-    Configuration,
     EnqueueLinksOptions,
     ErrorHandler,
     GetUserDataFromRequest,
     HttpCrawlerOptions,
     InternalHttpCrawlingContext,
     InternalHttpHook,
     RequestHandler,
-    RequestQueue,
     RouterRoutes,
+    Configuration,
+    RequestProvider,
+} from '@crawlee/http';
+import {
+    HttpCrawler,
+    enqueueLinks,
+    Router,
+    resolveBaseUrlForEnqueueLinksFiltering,
+    tryAbsoluteURL,
 } from '@crawlee/http';
-import { HttpCrawler, enqueueLinks, Router, resolveBaseUrlForEnqueueLinksFiltering, tryAbsoluteURL } from '@crawlee/http';
 import type { Dictionary } from '@crawlee/types';
 import type { CheerioOptions } from 'cheerio';
 import * as cheerio from 'cheerio';
@@ -203,7 +209,7 @@ export class CheerioCrawler extends HttpCrawler<CheerioCrawlingContext> {
 interface EnqueueLinksInternalOptions {
     options?: EnqueueLinksOptions;
     $: cheerio.CheerioAPI | null;
-    requestQueue: RequestQueue;
+    requestQueue: RequestProvider;
     originalRequestUrl: string;
     finalRequestUrl?: string;
 }

packages/core/src/enqueue_links/enqueue_links.ts

@@ -14,7 +14,7 @@ import {
     createRequests,
 } from './shared';
 import type { RequestOptions } from '../request';
-import type { RequestQueue, RequestQueueOperationOptions } from '../storages/request_queue';
+import type { RequestProvider, RequestQueueOperationOptions } from '../storages';
 
 export interface EnqueueLinksOptions extends RequestQueueOperationOptions {
     /** Limit the amount of actually enqueued URLs to this number. Useful for testing across the entire crawling scope. */
@@ -24,7 +24,7 @@ export interface EnqueueLinksOptions extends RequestQueueOperationOptions {
     urls?: string[];
 
     /** A request queue to which the URLs will be enqueued. */
-    requestQueue?: RequestQueue;
+    requestQueue?: RequestProvider;
 
     /** A CSS selector matching links to be enqueued. */
     selector?: string;

packages/core/src/request.ts

@@ -5,7 +5,7 @@ import util from 'node:util';
 import { normalizeUrl } from '@apify/utilities';
 import type { Dictionary } from '@crawlee/types';
 import type { BasePredicate } from 'ow';
-import ow, { ArgumentError } from 'ow';
+import ow from 'ow';
 
 import { log as defaultLog } from './log';
 import type { AllowedHttpMethods } from './typedefs';
@@ -141,14 +141,15 @@ export class Request<UserData extends Dictionary = Dictionary> {
         // properties and speeds up the validation cca 3-fold.
         // See https://github.com/sindresorhus/ow/issues/193
         keys(options).forEach((prop) => {
+            // skip url, because it is validated above
+            if (prop === 'url') {
+                return;
+            }
+
             const predicate = requestOptionalPredicates[prop as keyof typeof requestOptionalPredicates];
             const value = options[prop];
             if (predicate) {
                 ow(value, `RequestOptions.${prop}`, predicate as BasePredicate);
-                // 'url' is checked above because it's not optional
-            } else if (prop !== 'url') {
-                const msg = `Did not expect property \`${prop}\` to exist, got \`${value}\` in object \`RequestOptions\``;
-                throw new ArgumentError(msg, this.constructor);
             }
         });
 
@@ -479,6 +480,9 @@ export interface RequestOptions<UserData extends Dictionary = Dictionary> {
     /** @internal */
     handledAt?: string;
 
+    /** @internal */
+    lockExpiresAt?: Date;
+
 }
 
 export interface PushErrorMessageOptions {

packages/core/src/storages/index.ts

@@ -1,6 +1,8 @@
 export * from './dataset';
 export * from './key_value_store';
 export * from './request_list';
+export * from './request_provider';
 export * from './request_queue';
+export * from './request_queue_v2';
 export * from './storage_manager';
 export * from './utils';