diff --git a/docs/guides/request_loaders.mdx b/docs/guides/request_loaders.mdx index d2be9bd6c3e6..949643c7416e 100644 --- a/docs/guides/request_loaders.mdx +++ b/docs/guides/request_loaders.mdx @@ -26,6 +26,7 @@ The request loader abstractions are built around two interfaces and a couple of - `IRequestLoader`: The base interface for reading requests in a crawl. - `IRequestManager`: Extends `IRequestLoader` with write capabilities (adding and reclaiming requests). - `RequestManagerTandem`: Combines a read-only `IRequestLoader` with a writable `IRequestManager`. +- `ThrottlingRequestManager`: Wraps a writable `IRequestManager` and paces requests per domain. And the concrete request loader implementations: @@ -79,6 +80,8 @@ class SitemapRequestLoader class RequestManagerTandem +class ThrottlingRequestManager + %% ======================== %% Inheritance arrows %% ======================== @@ -88,6 +91,7 @@ IRequestLoader <|.. RequestList IRequestLoader <|.. SitemapRequestLoader IRequestManager <|.. RequestQueue IRequestManager <|.. RequestManagerTandem +IRequestManager <|.. ThrottlingRequestManager ``` :::info Crawler usage @@ -130,6 +134,42 @@ The loader supports filtering URLs using glob patterns and regular expressions, The `IRequestManager` interface extends `IRequestLoader` with **write** capabilities. In addition to reading requests, a request manager can add new requests and reclaim failed ones. This is essential for dynamic crawling, where new URLs emerge during the crawl, or when requests fail and need to be retried. The `RequestQueue` is the primary built-in request manager — see the [Request storage](./request-storage) guide for details. +## Per-domain throttling + +Some sites answer bursts of traffic with HTTP 429 (Too Many Requests) rather than an outright block. By default a 429 is treated as a blocked session: the session is retired and the request is retried straight away on a fresh one, which churns through proxies without actually slowing down. + +The `ThrottlingRequestManager` handles it at the scheduling layer instead. Wrap your request manager in it and list the domains you want paced: + +```ts +import { CheerioCrawler, RequestQueue, ThrottlingRequestManager } from 'crawlee'; + +const crawler = new CheerioCrawler({ + requestManager: new ThrottlingRequestManager({ + inner: await RequestQueue.open(), + domains: ['api.example.com'], + // optional, these are the defaults + baseDelaySecs: 2, + maxDelaySecs: 60, + maxDomainStallSecs: 900, + }), + requestHandler: async ({ request }) => { + // ... + }, +}); +``` + +Requests for a listed domain are routed into their own queue as they are added. When one of those domains answers with a 429, the crawler honours its `Retry-After` header — or backs off exponentially from `baseDelaySecs` up to `maxDelaySecs` if there is none — and holds that domain's requests back for the duration. Requests for every other domain keep flowing at full speed, the throttled request is retried later without counting against `maxRequestRetries`, and its session is left alone, because a rate limit says nothing about the session. + +Because a throttled request costs no retries, a domain that never stops rate-limiting would otherwise keep the crawl alive forever. If one goes `maxDomainStallSecs` without letting a single request through, the crawl shuts down with a `PersistentRateLimitError` — at that point the concurrency is too high for that domain, or it has blocked you outright, and waiting longer will not help. Its requests are left in their queue on purpose, so re-running the crawl without purging storages resumes them if the rate limit lifts. A crawler running with `keepAlive` is exempt, since staying up regardless is what it was asked to do. + +This is opt-in and exact: only the domains you list are throttled, and matching is case-insensitive with no wildcard support, so list each subdomain you care about. + +:::note robots.txt crawl-delay + +`ThrottlingRequestManager` is also what enforces `Crawl-delay` directives when `respectRobotsTxtFile` is enabled. Without it, or for a domain missing from `domains`, the directive is ignored and the crawler warns you about it. + +::: + ## Request manager tandem The `RequestManagerTandem` class combines the read-only capabilities of an `IRequestLoader` (like `RequestList`) with the read-write capabilities of an `IRequestManager` (like `RequestQueue`). This is useful when you need to load initial requests from a static source (such as a file, sitemap, or database) and also dynamically add or retry requests during the crawl. @@ -174,6 +214,6 @@ Similarly, you can combine a `Site ## Conclusion -This guide introduced the request loader abstractions: the read-only `IRequestLoader`, the writable `IRequestManager`, and the `RequestManagerTandem` that combines them, along with the `RequestList` and `SitemapRequestLoader` implementations. You also saw how to pair a loader with a queue using the `toTandem()` helper to handle both static and dynamically discovered requests. +This guide introduced the request loader abstractions: the read-only `IRequestLoader`, the writable `IRequestManager`, and the `RequestManagerTandem` that combines them, along with the `RequestList` and `SitemapRequestLoader` implementations. You also saw how to pair a loader with a queue using the `toTandem()` helper to handle both static and dynamically discovered requests, and how `ThrottlingRequestManager` paces requests to individual domains. If you have questions or need assistance, feel free to reach out on our [GitHub](https://github.com/apify/crawlee) or join our [Discord community](https://discord.com/invite/jyEM2PRvMU). Happy scraping! diff --git a/docs/guides/session_management.mdx b/docs/guides/session_management.mdx index 067e8a9c33de..46cd827037a5 100644 --- a/docs/guides/session_management.mdx +++ b/docs/guides/session_management.mdx @@ -213,6 +213,14 @@ const crawler = new CheerioCrawler({ For sites that respond with a `200` page that is actually a bot wall (Cloudflare challenges, Google's rate-limit page), set `retryOnBlocked: true` to have the crawler detect those by content and retry as well. For deeper anti-blocking measures see the [avoid blocking guide](./avoid-blocking). +:::tip A 429 is a rate limit, not a block + +Retiring a session on HTTP 429 burns proxies without slowing anything down — the site is asking you to wait, not telling you the session is unwelcome. Wrap your request manager in a `ThrottlingRequestManager` to back off per domain instead of rotating; see [per-domain throttling](./request-loaders#per-domain-throttling). Sessions are left untouched for the domains it covers. + +Those domains are handled as rate limits before `blockedStatusCodes` is consulted, so leave 429 in the list — removing it only changes what happens for domains the manager does not cover. + +::: + ## Sharing a session pool between crawlers A `SessionPool` instance can be shared across multiple crawlers by passing the same object to each crawler's `sessionPool` option. This is useful in multi-stage scrapers — for example a fast `CheerioCrawler` that discovers links and a `PlaywrightCrawler` that renders detail pages — where you want both stages to reuse the same proven, non-blocked identities and their cookies instead of each warming up its own pool from scratch. diff --git a/docs/public-api/crawlee-basic.api.md b/docs/public-api/crawlee-basic.api.md index 2bed167d88e7..0040bcbc045e 100644 --- a/docs/public-api/crawlee-basic.api.md +++ b/docs/public-api/crawlee-basic.api.md @@ -123,6 +123,7 @@ export class BasicCrawler & BasePredicate; requestList: ObjectPredicate & BasePredicate; requestQueue: ObjectPredicate & BasePredicate; + requestManager: ObjectPredicate & BasePredicate; requestHandler: Predicate & BasePredicate; requestHandlerTimeoutSecs: NumberPredicate & BasePredicate; errorHandler: Predicate & BasePredicate; @@ -159,6 +160,7 @@ export class BasicCrawler; readonly proxyConfiguration?: IProxyConfiguration; pushData(data: Parameters[0], datasetIdentifier?: string | StorageIdentifier): Promise; + protected recordDomainRateLimit(url: string, retryAfterHeader?: string | null): boolean; // (undocumented) protected readonly requestHandler: RequestHandler; protected requestManager?: IRequestManager; diff --git a/docs/public-api/crawlee-browser.api.md b/docs/public-api/crawlee-browser.api.md index a20e2fafe97c..9d202ad41375 100644 --- a/docs/public-api/crawlee-browser.api.md +++ b/docs/public-api/crawlee-browser.api.md @@ -41,6 +41,7 @@ import { StringPredicate } from 'ow'; // Not exported by the entry point; reachable only as a referenced type. // @public (undocumented) interface BaseResponse { + headers?(): Record; // (undocumented) status(): number; } @@ -79,6 +80,7 @@ export abstract class BrowserCrawler & BasePredicate; requestList: ObjectPredicate & BasePredicate; requestQueue: ObjectPredicate & BasePredicate; + requestManager: ObjectPredicate & BasePredicate; requestHandler: Predicate & BasePredicate; requestHandlerTimeoutSecs: NumberPredicate & BasePredicate; errorHandler: Predicate & BasePredicate; diff --git a/docs/public-api/crawlee-core.api.md b/docs/public-api/crawlee-core.api.md index 3447370f17d9..0342cd723f81 100644 --- a/docs/public-api/crawlee-core.api.md +++ b/docs/public-api/crawlee-core.api.md @@ -1149,6 +1149,9 @@ interface NewUrlOptions { export class NonRetryableError extends Error { } +// @public +export function parseRetryAfterHeader(value?: string | null): number | null; + // @public export function parseValue(body: Buffer | ArrayBuffer | string, contentTypeHeader: string | null): string | Buffer | ArrayBuffer | Record; @@ -1160,6 +1163,10 @@ export interface PersistenceOptions { enable?: boolean; } +// @public +export class PersistentRateLimitError extends CriticalError { +} + // @public export class ProxyConfiguration implements IProxyConfiguration { constructor(options?: ProxyConfigurationOptions); @@ -1336,6 +1343,9 @@ export interface RequestListState { nextUniqueKey: string | null; } +// @public +export type RequestManagerOpener = (identifier: string | StorageIdentifier, options?: StorageOpenOptions) => Promise; + // @public export class RequestManagerTandem implements IRequestManager { // (undocumented) @@ -1531,6 +1541,11 @@ export enum RequestState { UNPROCESSED = 0 } +// @public +export class RequestThrottledError extends RetryRequestError { + constructor(message?: string); +} + // @public export interface RequestTransform { // (undocumented) @@ -2086,6 +2101,19 @@ export interface StorageWritePolicy { requestQueue: StorageWriteMode; } +// @public +export interface SupportsDomainThrottling { + // (undocumented) + assertNoStalledDomains(): Promise; + // (undocumented) + recordDomainDelay(url: string, retryAfterMs?: number | null): boolean; + // (undocumented) + setCrawlDelay(url: string, delaySeconds: number): boolean; +} + +// @public +export function supportsDomainThrottling(manager: unknown): manager is SupportsDomainThrottling; + // @public export interface SystemInfo { // (undocumented) @@ -2110,6 +2138,50 @@ export interface TaskLoopPredicates { isTaskReadyFunction?: () => Promise; } +// @public +export class ThrottlingRequestManager implements IRequestManager, SupportsDomainThrottling { + // (undocumented) + [Symbol.asyncIterator](): AsyncGenerator, void, unknown>; + constructor(options: ThrottlingRequestManagerOptions, config?: Configuration); + // (undocumented) + addRequest(requestLike: Source, options?: RequestQueueOperationOptions): Promise; + addRequestsBatched(requests: RequestsLike, options?: AddRequestsBatchedOptions): Promise; + assertNoStalledDomains(): Promise; + // (undocumented) + drop(): Promise; + fetchNextRequest(): Promise | null>; + // (undocumented) + getHandledCount(): Promise; + // (undocumented) + getPendingCount(): Promise; + // (undocumented) + getTotalCount(): Promise; + get innerManager(): T; + isEmpty(): Promise; + isFinished(): Promise; + // (undocumented) + markRequestAsHandled(request: Request_2): Promise; + // (undocumented) + persistState(): Promise; + purge(): Promise; + // (undocumented) + reclaimRequest(request: Request_2, options?: RequestQueueOperationOptions): Promise; + recordDomainDelay(url: string, retryAfterMs?: number | null): boolean; + setCrawlDelay(url: string, delaySeconds: number): boolean; + // (undocumented) + setExpectedRequestProcessingTimeSecs(secs: number): Promise; +} + +// @public +export interface ThrottlingRequestManagerOptions { + baseDelaySecs?: number; + domains: string[]; + inner: T; + maxDelaySecs?: number; + maxDomainStallSecs?: number; + requestManagerOpener?: RequestManagerOpener; +} + export { tryAbsoluteURL } // @public diff --git a/docs/public-api/crawlee-http.api.md b/docs/public-api/crawlee-http.api.md index f8303c50ed4d..4fe3c6d45957 100644 --- a/docs/public-api/crawlee-http.api.md +++ b/docs/public-api/crawlee-http.api.md @@ -128,6 +128,7 @@ export class HttpCrawler = extendContext: Predicate & BasePredicate; requestList: ObjectPredicate & BasePredicate; requestQueue: ObjectPredicate & BasePredicate; + requestManager: ObjectPredicate & BasePredicate; requestHandler: Predicate & BasePredicate; requestHandlerTimeoutSecs: NumberPredicate & BasePredicate; errorHandler: Predicate & BasePredicate; diff --git a/docs/public-api/crawlee-jsdom.api.md b/docs/public-api/crawlee-jsdom.api.md index 5edf593b5df6..2338e4fb93ac 100644 --- a/docs/public-api/crawlee-jsdom.api.md +++ b/docs/public-api/crawlee-jsdom.api.md @@ -72,6 +72,7 @@ export class JSDOMCrawler, ExtendedContext extendContext: Predicate & BasePredicate; requestList: ObjectPredicate & BasePredicate; requestQueue: ObjectPredicate & BasePredicate; + requestManager: ObjectPredicate & BasePredicate; requestHandler: Predicate & BasePredicate; requestHandlerTimeoutSecs: NumberPredicate & BasePredicate; errorHandler: Predicate & BasePredicate; diff --git a/docs/public-api/crawlee-playwright.api.md b/docs/public-api/crawlee-playwright.api.md index 48b3b21d6296..686e7a6562d7 100644 --- a/docs/public-api/crawlee-playwright.api.md +++ b/docs/public-api/crawlee-playwright.api.md @@ -361,6 +361,7 @@ export class PlaywrightCrawler, ExtendedCon extendContext: Predicate & BasePredicate; requestList: ObjectPredicate & BasePredicate; requestQueue: ObjectPredicate & BasePredicate; + requestManager: ObjectPredicate & BasePredicate; requestHandler: Predicate & BasePredicate; requestHandlerTimeoutSecs: NumberPredicate & BasePredicate; errorHandler: Predicate & BasePredicate; diff --git a/docs/public-api/crawlee-puppeteer.api.md b/docs/public-api/crawlee-puppeteer.api.md index 54969e9e27dd..0d848e243475 100644 --- a/docs/public-api/crawlee-puppeteer.api.md +++ b/docs/public-api/crawlee-puppeteer.api.md @@ -215,6 +215,7 @@ export class PuppeteerCrawler, ExtendedCont extendContext: Predicate & BasePredicate; requestList: ObjectPredicate & BasePredicate; requestQueue: ObjectPredicate & BasePredicate; + requestManager: ObjectPredicate & BasePredicate; requestHandler: Predicate & BasePredicate; requestHandlerTimeoutSecs: NumberPredicate & BasePredicate; errorHandler: Predicate & BasePredicate; diff --git a/docs/public-api/crawlee-stagehand.api.md b/docs/public-api/crawlee-stagehand.api.md index 84b28934ddc4..bd340377c92e 100644 --- a/docs/public-api/crawlee-stagehand.api.md +++ b/docs/public-api/crawlee-stagehand.api.md @@ -103,6 +103,7 @@ export class StagehandCrawler, ExtendedCont extendContext: Predicate & BasePredicate; requestList: ObjectPredicate & BasePredicate; requestQueue: ObjectPredicate & BasePredicate; + requestManager: ObjectPredicate & BasePredicate; requestHandler: Predicate & BasePredicate; requestHandlerTimeoutSecs: NumberPredicate & BasePredicate; errorHandler: Predicate & BasePredicate; diff --git a/docs/public-api/crawlee-utils.api.md b/docs/public-api/crawlee-utils.api.md index ff0697a802c2..a81cdb1ca43c 100644 --- a/docs/public-api/crawlee-utils.api.md +++ b/docs/public-api/crawlee-utils.api.md @@ -152,6 +152,7 @@ export class RobotsTxtFile { logger?: CrawleeLogger; }): Promise; static from(url: string, content: string, proxyUrl?: string): RobotsTxtFile; + getCrawlDelay(userAgent?: string): number | undefined; getSitemaps(): string[]; isAllowed(url: string, userAgent?: string): boolean; parseSitemaps(): Promise; diff --git a/docs/upgrading/upgrading_v4.md b/docs/upgrading/upgrading_v4.md index a537dd5d4fb3..e8fbf8540877 100644 --- a/docs/upgrading/upgrading_v4.md +++ b/docs/upgrading/upgrading_v4.md @@ -1174,6 +1174,24 @@ const crawler = new CheerioCrawler({ requestManager: await requestList.toTandem( A lone `requestList` now runs through a tandem over an auto-opened queue (rather than a read-only adapter). This means retries and `maxRequestsPerCrawl` accounting for that path now follow queue semantics. +#### HTTP 429 can now back off per domain instead of retiring the session + +`blockedStatusCodes` still defaults to `[401, 403, 429]`, so out of the box a 429 retires the session and retries immediately, as in v3. New in v4 is the opt-in `ThrottlingRequestManager`, which handles rate limits at the scheduling layer instead: + +```typescript +const crawler = new CheerioCrawler({ + requestManager: new ThrottlingRequestManager({ + inner: await RequestQueue.open(), + domains: ['api.example.com'], + }), + requestHandler, +}); +``` + +For the domains you list, a 429 is treated as a rate limit before `blockedStatusCodes` is consulted at all — it honours `Retry-After` (or backs off exponentially), holds only that domain's requests back, and leaves both the session and the request's retry budget untouched. Removing 429 from `blockedStatusCodes` therefore only affects domains the manager does not cover; you do not need to touch it to adopt throttling. Because those retries are free, a domain that never stops rate-limiting would keep the crawl alive indefinitely — so one that goes `maxDomainStallSecs` (15 minutes by default) without letting a single request through shuts the crawl down with a `PersistentRateLimitError`, leaving its requests queued for a later run — unless `keepAlive` is set, which exempts the crawl. + +It is also what enforces robots.txt `Crawl-delay` directives — with `respectRobotsTxtFile` enabled and no throttling manager covering the domain, the directive is ignored and the crawler warns about it. See the [request loaders guide](../guides/request-loaders#per-domain-throttling). + #### `BasicCrawler.requestList` and `BasicCrawler.requestQueue` fields removed The public `requestList` and `requestQueue` instance fields are gone. The crawler exposes a single `protected requestManager?: IRequestManager` instead. Access the active manager via the new async `getRequestManager()` method. diff --git a/packages/basic-crawler/src/internals/basic-crawler.ts b/packages/basic-crawler/src/internals/basic-crawler.ts index 27130f6fa25b..b93244b61cce 100644 --- a/packages/basic-crawler/src/internals/basic-crawler.ts +++ b/packages/basic-crawler/src/internals/basic-crawler.ts @@ -60,10 +60,13 @@ import { OwnedOrInjected, purgeDefaultStorages, RequestHandlerError, + parseRetryAfterHeader, + RequestThrottledError, RequestManagerTandem, RequestQueue, RequestState, RetryRequestError, + supportsDomainThrottling, Router, ServiceLocator, serviceLocator, @@ -420,6 +423,10 @@ export interface BasicCrawlerOptions< /** * HTTP status codes that indicate the session should be retired. + * + * A 429 from a domain covered by a {@apilink ThrottlingRequestManager} is handled as a rate limit before + * this is consulted, so removing 429 here only affects domains that manager does not cover. + * * @default [401, 403, 429] */ blockedStatusCodes?: number[]; @@ -827,6 +834,7 @@ export class BasicCrawler< requestList: ow.optional.object.validate(validators.requestList), requestQueue: ow.optional.object.validate(validators.requestQueue), + requestManager: ow.optional.object, // Subclasses override this function instead of passing it // in constructor, so this validation needs to apply only // if the user creates an instance of BasicCrawler directly. @@ -1163,7 +1171,7 @@ export class BasicCrawler< ); // SessionError already retired the session in `requestFunctionErrorHandler`; // skip `markBad` to avoid double-counting usage/error score. - if (!(unwrappedError instanceof SessionError)) { + if (!this.errorAbsolvesSession(unwrappedError)) { crawlingContext.session?.markBad(); } return; @@ -1222,6 +1230,12 @@ export class BasicCrawler< return true; } + // Checked here because this runs only once nothing is in flight, which is exactly when a + // crawl that cannot progress looks indistinguishable from one that is merely waiting. + if (!keepAlive && supportsDomainThrottling(this.requestManager)) { + await this.requestManager.assertNoStalledDomains(); + } + const isFinished = isFinishedFunction ? await isFinishedFunction() : await this.defaultIsFinishedFunction(); @@ -1882,9 +1896,9 @@ export class BasicCrawler< }); } - private logOncePerRun(key: string, message: string): void { + private logOncePerRun(key: string, message: string, level: 'info' | 'warning' = 'info'): void { if (!this.#loggedPerRun.has(key)) { - this.log.info(message); + this.log[level](message); this.#loggedPerRun.add(key); } } @@ -2226,9 +2240,65 @@ export class BasicCrawler< const robotsTxtFile = await this.getRobotsTxtFileForUrl(url); const userAgent = typeof this.#respectRobotsTxtFile === 'object' ? this.#respectRobotsTxtFile?.userAgent : '*'; + if (robotsTxtFile) { + const crawlDelay = robotsTxtFile.getCrawlDelay(userAgent); + if (crawlDelay !== undefined) { + this.applyCrawlDelay(url, crawlDelay); + } + } + return !robotsTxtFile || robotsTxtFile.isAllowed(url, userAgent); } + /** + * Records an HTTP 429 against the URL's domain so the request manager can pace the retry. + * + * @param retryAfterHeader The raw `Retry-After` response header, if the server sent one. + * @returns `true` if a manager took responsibility for the delay, in which case the caller should throw + * {@apilink RequestThrottledError} rather than treating the response as a blocked session. + */ + protected recordDomainRateLimit(url: string, retryAfterHeader?: string | null): boolean { + if ( + supportsDomainThrottling(this.requestManager) && + this.requestManager.recordDomainDelay(url, parseRetryAfterHeader(retryAfterHeader)) + ) { + return true; + } + + const domain = hostnameOrUrl(url); + this.logOncePerRun( + `rateLimitNotThrottled:${domain}`, + `"${domain}" responded with HTTP 429 (Too Many Requests), but nothing is set up to back off from it, ` + + 'so the response is handled like any other, with no per-domain delay. ' + + `Pass a \`ThrottlingRequestManager\` as \`requestManager\` and include "${domain}" in its \`domains\` ` + + 'option to honour `Retry-After` and apply exponential backoff instead.', + 'warning', + ); + + return false; + } + + /** + * Hands a robots.txt `Crawl-delay` to the request manager, warning if nothing is able to honour it. + * + * The warning is driven by whether the delay was actually accepted rather than by the type of the manager, + * because a manager that does throttle still drops the delay for a domain missing from its `domains` list. + */ + private applyCrawlDelay(url: string, delaySeconds: number): void { + if (supportsDomainThrottling(this.requestManager) && this.requestManager.setCrawlDelay(url, delaySeconds)) { + return; + } + + const domain = hostnameOrUrl(url); + this.logOncePerRun( + `crawlDelayIgnored:${domain}`, + `robots.txt for "${domain}" defines a crawl-delay of ${delaySeconds}s, but nothing is set up to honour it, ` + + 'so requests to that domain will not be paced. Pass a `ThrottlingRequestManager` as `requestManager` ' + + `and include "${domain}" in its \`domains\` option to enforce the delay.`, + 'warning', + ); + } + protected async getRobotsTxtFileForUrl(url: string): Promise { if (!this.#respectRobotsTxtFile) { return undefined; @@ -2417,7 +2487,7 @@ export class BasicCrawler< } // decrease the session score if the request fails (but the error handler did not throw); // skip when the error is a SessionError, which already retired the session - if (!(err instanceof SessionError)) { + if (!this.errorAbsolvesSession(err)) { crawlingContext.session.markBad(); } } finally { @@ -2577,6 +2647,17 @@ export class BasicCrawler< request: Request, source: IRequestManager, ): Promise { + if (error instanceof RequestThrottledError) { + // The domain told us to come back later, so the request was never really attempted. Put it back + // without recording a failure - it costs neither a retry nor session reputation. + this.log.debug(`Deferring request because its domain is rate-limiting us. ${error.message}`, { + id: request.id, + url: request.url, + }); + await source.reclaimRequest(request, { forefront: request.userData?.__crawlee?.forefront }); + return; + } + request.pushErrorMessage(error); if (error instanceof CriticalError) { @@ -2677,6 +2758,14 @@ export class BasicCrawler< : [error.message || error, userLine].join('\n'); } + /** + * Whether the session should be spared for this error - either because it was already retired, or because the + * failure says nothing about the session (a rate limit is a property of the domain). + */ + private errorAbsolvesSession(error: Error): boolean { + return error instanceof SessionError || error instanceof RequestThrottledError; + } + private canRequestBeRetried(request: Request, error: Error) { // Request should never be retried, or the error encountered makes it not able to be retried. if (request.noRetry || error instanceof NonRetryableError) { @@ -2807,6 +2896,11 @@ export interface CrawlerRunOptions extends CrawlerAddRequestsOptions { purgeRequestQueue?: boolean; } +/** The hostname of `url`, falling back to the whole string when it is not parseable - for log messages only. */ +function hostnameOrUrl(url: string): string { + return URL.canParse(url) ? new URL(url).hostname : url; +} + /** * Creates new {@apilink Router} instance that works based on request labels. * This instance can then serve as a {@apilink BasicCrawlerOptions.requestHandler|`requestHandler`} of our {@apilink BasicCrawler}. diff --git a/packages/browser-crawler/src/internals/browser-crawler.ts b/packages/browser-crawler/src/internals/browser-crawler.ts index 6ba1c25afdd0..a14b276ca17c 100644 --- a/packages/browser-crawler/src/internals/browser-crawler.ts +++ b/packages/browser-crawler/src/internals/browser-crawler.ts @@ -23,6 +23,7 @@ import { OwnedOrInjected, remainingNavigationWindowMillis, RequestState, + RequestThrottledError, resolveBaseUrlForEnqueueLinksFiltering, SessionError, toughCookieToBrowserPoolCookie, @@ -60,6 +61,8 @@ import type { BrowserLaunchContext } from './browser-launcher.js'; interface BaseResponse { status(): number; + /** Optional because only Playwright and Puppeteer responses are guaranteed to carry it. */ + headers?(): Record; } /** @@ -832,6 +835,16 @@ export abstract class BrowserCrawler< this.stats.registerStatusCode(status); + // Ahead of the error-status throw below: a 429 the user opted into treating as an error is still a + // rate limit the domain should back off from. + if (status === 429) { + // Both drivers lower-case header names and join duplicates, so a plain lookup is enough. + const retryAfter = response.headers?.()['retry-after']; + if (this.recordDomainRateLimit(crawlingContext.request.url, retryAfter)) { + throw new RequestThrottledError(`${crawlingContext.request.url} responded with 429.`); + } + } + if (this.isErrorStatusCode(status)) { if (this.additionalHttpErrorStatusCodes.has(status)) { throw new Error(`${status} - Error status code was set by user.`); diff --git a/packages/core/src/errors.ts b/packages/core/src/errors.ts index 13a1c79b3ee4..a5ed96dfcb5c 100644 --- a/packages/core/src/errors.ts +++ b/packages/core/src/errors.ts @@ -53,6 +53,29 @@ export class RetryRequestError extends Error { } } +/** + * Thrown when a domain has rate-limited us and the request should simply be attempted again later. + * + * The request is reclaimed without recording a failure: it costs neither a retry nor session reputation, because + * nothing about the request or the session was at fault. A {@apilink ThrottlingRequestManager} holds it back until + * the domain's backoff expires, so retries are paced rather than immediate. + */ +export class RequestThrottledError extends RetryRequestError { + constructor(message?: string) { + super(message ?? 'Request is being retried later because its domain is rate-limiting us'); + } +} + +/** + * Thrown when a domain has rate-limited us for so long that no request has got through, and the crawl is + * abandoned rather than kept waiting. + * + * Waiting longer will not help: at this point the concurrency is too high for the domain, or it has blocked us. + * The affected requests are deliberately left in their queue, so re-running the crawl without purging storages + * resumes them once the domain recovers. + */ +export class PersistentRateLimitError extends CriticalError {} + /** * Errors of `SessionError` type retire the session associated with the request and trigger a regular retry. * diff --git a/packages/core/src/http.ts b/packages/core/src/http.ts new file mode 100644 index 000000000000..4f3696d46ce1 --- /dev/null +++ b/packages/core/src/http.ts @@ -0,0 +1,32 @@ +/** + * Parses a `Retry-After` response header into a delay in milliseconds. + * + * The header holds either a non-negative number of seconds or an HTTP-date. + * See [MDN](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Retry-After). + * + * @returns The delay in milliseconds, or `null` if the header is absent, unparseable, or already elapsed. + */ +export function parseRetryAfterHeader(value?: string | null): number | null { + if (!value) { + return null; + } + + const trimmed = value.trim(); + + // Per the spec this is a `delay-seconds`: digits only, so a negative or fractional value is not one. + if (/^\d+$/.test(trimmed)) { + // `Retry-After: 0` names no future deadline, same as an HTTP-date that has already passed. Reporting it + // as a zero delay would leave the domain unthrottled while still counting as a rate-limit event, so the + // caller would defer the request for free and re-send it immediately. + const delayMs = Number(trimmed) * 1000; + return delayMs > 0 ? delayMs : null; + } + + const date = Date.parse(trimmed); + if (!Number.isNaN(date)) { + const delayMs = date - Date.now(); + return delayMs > 0 ? delayMs : null; + } + + return null; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index dcf7df4a57d7..29a05a552e49 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -17,5 +17,6 @@ export * from './storages/index.js'; export * from './memory-storage/index.js'; export * from './validators.js'; export * from './cookie_utils.js'; +export * from './http.js'; export * from './recoverable_state.js'; export type { StorageBackend } from '@crawlee/types'; diff --git a/packages/core/src/storages/batched_adds.ts b/packages/core/src/storages/batched_adds.ts new file mode 100644 index 000000000000..ecc51ffb8aa6 --- /dev/null +++ b/packages/core/src/storages/batched_adds.ts @@ -0,0 +1,136 @@ +import { setTimeout as sleep } from 'node:timers/promises'; +import type { ProcessedRequest } from '@crawlee/types'; + +import { chunkedAsyncIterable, peekableAsyncIterable } from '../iterables.js'; +import type { Source } from '../request.js'; +import type { AddRequestsBatchedResult } from './request_queue.js'; +import { activeStorageTransaction, withDirectStorageAccess } from './transaction.js'; + +export interface DrainRequestBatchesOptions { + /** + * The requests to add, already normalized by the caller. Consumed lazily: an unbounded or expensive + * iterable is only pulled from as far as the batching (and any `maxNewRequests` budget) requires. + */ + items: AsyncGenerator; + + batchSize: number; + waitBetweenBatchesMillis: number; + waitForAllRequestsToBeAdded: boolean; + maxNewRequests?: number; + + /** + * Adds a single chunk and reports what it processed. + * + * @param isInitial Whether this is the first chunk, which is added before this function returns. Later + * chunks land in the background, which is why some callers cache only the first. + */ + processChunk: (chunk: TItem[], isInitial: boolean) => Promise; + + /** + * Called with the promise covering every chunk after the first, so the caller can keep its own + * `isFinished` honest while batches are still landing. + */ + trackBackgroundBatches?: (batches: Promise) => void; +} + +/** + * Drives the chunk-by-chunk half of `addRequestsBatched`: the first chunk is added before returning and the + * rest continue in the background, paced by `waitBetweenBatchesMillis`. + * + * Callers differ only in how a chunk is added and how the input is normalized, so that is all + * {@apilink DrainRequestBatchesOptions} asks for - the budget arithmetic, the lazy chunking, the + * over-limit reporting and the transaction handling are identical for everyone and live here. In + * particular, every caller has to keep its background chunks out of a transaction they will outlive, + * so that is read from the ambient transaction rather than asked of the caller. + */ +export async function drainRequestBatches( + options: DrainRequestBatchesOptions, +): Promise { + const { + items, + batchSize, + waitBetweenBatchesMillis, + waitForAllRequestsToBeAdded, + maxNewRequests, + processChunk, + trackBackgroundBatches, + } = options; + + const deferred = activeStorageTransaction()?.policy.requestQueue === 'deferred'; + + let remainingBudget = maxNewRequests ?? Infinity; + const requestsOverLimit: Source[] = []; + + // Never hand a chunk more than the budget allows, so an over-large final batch cannot overshoot. + const effectiveChunkSize = maxNewRequests !== undefined ? () => Math.min(batchSize, remainingBudget) : batchSize; + + const chunks = peekableAsyncIterable(chunkedAsyncIterable(items, effectiveChunkSize)); + const chunksIterator = chunks[Symbol.asyncIterator](); + + const addChunk = async (chunk: TItem[], isInitial: boolean) => { + const processedRequests = await processChunk(chunk, isInitial); + + if (maxNewRequests !== undefined) { + remainingBudget -= processedRequests.filter((request) => !request.wasAlreadyPresent).length; + } + + return processedRequests; + }; + + const buildResult = async ( + addedRequests: ProcessedRequest[], + waitForAll: Promise, + ): Promise => { + if (maxNewRequests !== undefined) { + // `chunkedAsyncIterable` stops pulling once the budget-derived chunk size hits zero, so whatever + // is left is still sitting in `items` rather than in a chunk we have seen. + for await (const item of items) { + requestsOverLimit.push(item); + } + } + + return { addedRequests, waitForAllRequestsToBeAdded: waitForAll, requestsOverLimit }; + }; + + const initialChunk = await chunksIterator.peek(); + if (initialChunk === undefined) { + return buildResult([], Promise.resolve([])); + } + + const addedRequests = await addChunk(initialChunk, true); + await chunksIterator.next(); + + if ((await chunksIterator.peek()) === undefined) { + return buildResult(addedRequests, Promise.resolve([])); + } + + const processRemainingChunks = async () => { + const added: ProcessedRequest[] = []; + for await (const chunk of chunks) { + added.push(...(await addChunk(chunk, false))); + // Under `deferred` no chunk performs backend I/O, so pacing them would only stall the handler. + await sleep(deferred ? 0 : waitBetweenBatchesMillis); + } + return added; + }; + + // With a budget we must drain everything before we can report what went over it; under `deferred` a + // writer that finishes after commit would have nowhere to put its journal entries. + const awaitsRemainder = waitForAllRequestsToBeAdded || maxNewRequests !== undefined || deferred; + + // An un-awaited writer outlives the transaction scope it inherits, so it must not record into a + // transaction that may already be closed. It writes directly - its write-through additions were never + // going to be rolled back anyway - which means the requests it adds are not journaled. + // See `StorageTransactionView.enqueuedUrls`. + const remainder = awaitsRemainder ? processRemainingChunks() : withDirectStorageAccess(processRemainingChunks); + + // The caller is not obliged to await `remainder`, so give it a handler of its own - an unhandled + // rejection here would otherwise take the process down. + trackBackgroundBatches?.(remainder.catch(() => {})); + + if (awaitsRemainder) { + addedRequests.push(...(await remainder)); + } + + return buildResult(addedRequests, remainder); +} diff --git a/packages/core/src/storages/index.ts b/packages/core/src/storages/index.ts index af38af4258ab..a59913f7f634 100644 --- a/packages/core/src/storages/index.ts +++ b/packages/core/src/storages/index.ts @@ -11,3 +11,4 @@ export * from './utils.js'; export * from './transaction.js'; export * from './sitemap_request_loader.js'; export * from './request_manager_tandem.js'; +export * from './throttling_request_manager.js'; diff --git a/packages/core/src/storages/request_loader.ts b/packages/core/src/storages/request_loader.ts index 32687f6c929c..f68988b35b72 100644 --- a/packages/core/src/storages/request_loader.ts +++ b/packages/core/src/storages/request_loader.ts @@ -60,6 +60,11 @@ export interface IRequestLoader { * Resolves to `true` if the next call to {@apilink IRequestLoader.fetchNextRequest} function * would return `null`, otherwise it resolves to `false`. * Note that even if the loader is empty, there might be some pending requests currently being processed. + * + * This is a statement about what the *next fetch* would return, not about how much work is left, so it + * may report `true` while {@apilink IRequestLoader.getPendingCount} is non-zero - a loader that withholds + * requests for a while (as {@apilink ThrottlingRequestManager} does for a rate-limited domain) is empty + * for as long as it will not hand anything over. Use `isFinished()` to ask whether the work is done. */ isEmpty(): Promise; diff --git a/packages/core/src/storages/request_queue.ts b/packages/core/src/storages/request_queue.ts index fbf0f3a3002b..df28b5b55c8a 100644 --- a/packages/core/src/storages/request_queue.ts +++ b/packages/core/src/storages/request_queue.ts @@ -11,7 +11,7 @@ import type { RequestQueueInfo, } from '@crawlee/types'; import { isAsyncIterable, isIterable } from '@crawlee/utils/internal'; -import { downloadListOfUrls, sleep } from '@crawlee/utils'; +import { downloadListOfUrls } from '@crawlee/utils'; import ow from 'ow'; import type { ReadonlyDeep } from 'type-fest'; @@ -22,14 +22,14 @@ import { Configuration } from '../configuration.js'; import { getObjectType } from '../debug.js'; import type { EventManager } from '../events/event_manager.js'; import { EventType } from '../events/event_manager.js'; -import { chunkedAsyncIterable, peekableAsyncIterable } from '../iterables.js'; import type { CrawleeLogger } from '../log.js'; import type { IProxyConfiguration } from '../proxy_configuration.js'; import type { InternalSource, RequestOptions, Source } from '../request.js'; import { Request } from '../request.js'; import { serviceLocator } from '../service_locator.js'; import type { JournalEntry, StorageTransaction } from './transaction.js'; -import { activeStorageTransaction, rejectOperationInTransaction, withDirectStorageAccess } from './transaction.js'; +import { activeStorageTransaction, rejectOperationInTransaction } from './transaction.js'; +import { drainRequestBatches } from './batched_adds.js'; import type { IRequestManager, RequestsLike } from './request_manager.js'; import type { RequestQueueStats } from './storage_stats.js'; import { StorageStatsTracker } from './storage_stats.js'; @@ -571,9 +571,6 @@ export class RequestQueue implements IStorage, IRequestManager { requests: ReadonlyDeep, options: AddRequestsBatchedOptions = {}, ): Promise { - const transaction = activeStorageTransaction(); - const deferred = transaction?.policy.requestQueue === 'deferred'; - ow( requests, ow.object @@ -630,133 +627,43 @@ export class RequestQueue implements IStorage, IRequestManager { } } - const { batchSize = 1000, maxNewRequests = undefined } = options; - // Under `deferred` no chunk performs backend I/O, so pacing them would only stall the handler. - const waitBetweenBatchesMillis = deferred ? 0 : (options.waitBetweenBatchesMillis ?? 1000); - - let remainingBudget = maxNewRequests ?? Infinity; - const requestsOverLimit: Source[] = []; - - // If there's a limit on the number of added requests, do not send batches bigger than the limit - const effectiveChunkSize = - maxNewRequests !== undefined ? () => Math.min(batchSize, remainingBudget) : batchSize; - - // Hold onto the underlying iterator so we can drain leftovers from it in buildResult - const requestIterator = generateRequests(); - - const chunks = peekableAsyncIterable( - chunkedAsyncIterable(requestIterator, effectiveChunkSize) as AsyncIterable, - ); - const chunksIterator = chunks[Symbol.asyncIterator](); - - /** - * Process a chunk: send it to the queue, then update the remaining budget if maxNewRequests is active. - * - * Requests the backend reports as unprocessed are warned about and skipped rather than retried: - * `unprocessedRequests` is what remains after the backend's own transient-error handling - a - * semantic rejection (e.g. a malformed `userData` shape) that re-sending would only re-poke. - * Retrying transient failures is the storage backend's job, not the frontend's. - */ - const processChunk = async (chunk: Source[], cache = true) => { - const { processedRequests, unprocessedRequests } = await this.addRequests(chunk, { - forefront: options.forefront, - cache, - }); - - if (unprocessedRequests.length > 0) { - this.log.warning( - 'Some requests were rejected by the request queue and will be skipped. ' + - "This usually means the request data is malformed (e.g. an invalid 'userData' shape).", - { unprocessedRequests }, - ); - } - - if (maxNewRequests !== undefined) { - remainingBudget -= processedRequests.filter((r) => !r.wasAlreadyPresent).length; - } - - return processedRequests; - }; + return drainRequestBatches({ + items: generateRequests(), + batchSize: options.batchSize ?? 1000, + waitBetweenBatchesMillis: options.waitBetweenBatchesMillis ?? 1000, + waitForAllRequestsToBeAdded: options.waitForAllRequestsToBeAdded ?? false, + maxNewRequests: options.maxNewRequests, + + /** + * Requests the backend reports as unprocessed are warned about and skipped rather than retried: + * `unprocessedRequests` is what remains after the backend's own transient-error handling - a + * semantic rejection (e.g. a malformed `userData` shape) that re-sending would only re-poke. + * Retrying transient failures is the storage backend's job, not the frontend's. + */ + processChunk: async (chunk, isInitial) => { + const { processedRequests, unprocessedRequests } = await this.addRequests(chunk, { + forefront: options.forefront, + cache: isInitial, + }); - /** - * Build the final result. When maxNewRequests is set, drains any remaining items - * from the underlying request iterator into requestsOverLimit. - * - * We accept the iterator explicitly (rather than closing over it) to make it obvious - * that this is the *same* iterator that `chunkedAsyncIterable` has been consuming — - * so only unconsumed items are drained. We drain `requestIterator` (not `chunks`) - * because `chunkedAsyncIterable` stops yielding when the budget-based chunk size - * drops to 0, leaving unconsumed items in the underlying iterator. - */ - const buildResult = async ( - addedRequests: ProcessedRequest[], - waitForAllRequestsToBeAdded: Promise, - unconsumedIterator: AsyncGenerator, - ): Promise => { - if (maxNewRequests !== undefined) { - for await (const request of unconsumedIterator) { - requestsOverLimit.push(request); + if (unprocessedRequests.length > 0) { + this.log.warning( + 'Some requests were rejected by the request queue and will be skipped. ' + + "This usually means the request data is malformed (e.g. an invalid 'userData' shape).", + { unprocessedRequests }, + ); } - } - - return { addedRequests, waitForAllRequestsToBeAdded, requestsOverLimit }; - }; - - // Add initial batch to process right away - const initialChunk = await chunksIterator.peek(); - if (initialChunk === undefined) { - return buildResult([], Promise.resolve([]), requestIterator); - } - - const addedRequests = await processChunk(initialChunk); - await chunksIterator.next(); - - // If we have no more requests to add (either exhausted or budget hit), return immediately - if ((await chunksIterator.peek()) === undefined) { - return buildResult(addedRequests, Promise.resolve([]), requestIterator); - } - - const processRemainingChunks = async () => { - const finalAddedRequests: ProcessedRequest[] = []; - - for await (const requestChunk of chunks) { - finalAddedRequests.push(...(await processChunk(requestChunk, false))); - await sleep(waitBetweenBatchesMillis); - } - - return finalAddedRequests; - }; - // maxNewRequests needs all batches to report skipped requests accurately; `deferred` needs them - // too - a writer that finishes after commit would have nowhere to put its journal entries. - const awaitsRemainingChunks = options.waitForAllRequestsToBeAdded || maxNewRequests !== undefined || deferred; - - // eslint-disable-next-line no-async-promise-executor - const promise = new Promise(async (resolve) => { - if (awaitsRemainingChunks) { - // Awaited below, i.e. still within the caller's transaction scope, so the additions are - // journaled like the initial chunk - introspection must not depend on where the chunk - // boundary happened to fall. - resolve(await processRemainingChunks()); - } else { - // Nobody awaits this writer, so it outlives the transaction scope it inherits and must - // not record into a transaction that may already be closed. It writes directly - its - // write-through additions were never going to be rolled back anyway - which means the - // requests it adds are not journaled. See `StorageTransactionView.enqueuedUrls`. - resolve(await withDirectStorageAccess(processRemainingChunks)); - } - }); + return processedRequests; + }, - this.inProgressRequestBatchCount += 1; - void promise.finally(() => { - this.inProgressRequestBatchCount -= 1; + trackBackgroundBatches: (batches) => { + this.inProgressRequestBatchCount += 1; + void batches.finally(() => { + this.inProgressRequestBatchCount -= 1; + }); + }, }); - - if (awaitsRemainingChunks) { - addedRequests.push(...(await promise)); - } - - return buildResult(addedRequests, promise, requestIterator); } /** diff --git a/packages/core/src/storages/throttling_request_manager.ts b/packages/core/src/storages/throttling_request_manager.ts new file mode 100644 index 000000000000..6db5f7efc206 --- /dev/null +++ b/packages/core/src/storages/throttling_request_manager.ts @@ -0,0 +1,688 @@ +import { URL } from 'node:url'; +import type { Dictionary } from '@crawlee/types'; +import ow from 'ow'; + +import type { Configuration } from '../configuration.js'; +import { PersistentRateLimitError } from '../errors.js'; +import { asyncifyIterable } from '../iterables.js'; +import type { CrawleeLogger } from '../log.js'; +import type { Request, Source } from '../request.js'; +import { serviceLocator } from '../service_locator.js'; +import { normalizeHostname } from '../url.js'; +import { drainRequestBatches } from './batched_adds.js'; +import type { IRequestManager, RequestsLike } from './request_manager.js'; +import type { + AddRequestsBatchedOptions, + AddRequestsBatchedResult, + RequestQueueOperationInfo, + RequestQueueOperationOptions, +} from './request_queue.js'; +import { RequestQueue } from './request_queue.js'; +import type { StorageIdentifier } from './storage_instance_manager.js'; +import type { StorageOpenOptions } from './utils.js'; + +/** + * Opens a request manager, matching the shape of storage `open` methods such as + * {@apilink RequestQueue.open|`RequestQueue.open`}. + * + * {@apilink ThrottlingRequestManager} calls this once per configured domain, so every per-domain queue shares the + * concrete type and storage backend of the manager being wrapped. + */ +export type RequestManagerOpener = ( + identifier: string | StorageIdentifier, + options?: StorageOpenOptions, +) => Promise; + +/** + * A request manager that can pace requests per domain, as {@apilink ThrottlingRequestManager} does. + * + * The crawlers detect this structurally rather than by type, so a wrapper can opt in by forwarding these three + * methods without {@apilink IRequestManager} having to know that throttling exists. + */ +export interface SupportsDomainThrottling { + /** @see {@apilink ThrottlingRequestManager.recordDomainDelay} */ + recordDomainDelay(url: string, retryAfterMs?: number | null): boolean; + /** @see {@apilink ThrottlingRequestManager.setCrawlDelay} */ + setCrawlDelay(url: string, delaySeconds: number): boolean; + /** @see {@apilink ThrottlingRequestManager.assertNoStalledDomains} */ + assertNoStalledDomains(): Promise; +} + +/** Whether `manager` can pace requests per domain. */ +export function supportsDomainThrottling(manager: unknown): manager is SupportsDomainThrottling { + const candidate = manager as Partial | null | undefined; + + return ( + typeof candidate?.recordDomainDelay === 'function' && + typeof candidate.setCrawlDelay === 'function' && + typeof candidate.assertNoStalledDomains === 'function' + ); +} + +/** Options for {@apilink ThrottlingRequestManager}. */ +export interface ThrottlingRequestManagerOptions { + /** + * The request manager to wrap, usually a {@apilink RequestQueue}. Requests for domains that are not throttled + * are stored here. + */ + inner: T; + + /** + * Hostnames to throttle. Matching is case-insensitive and exact - wildcards such as `*.example.com` are not + * supported, so list each subdomain you care about. Requests for any other domain bypass throttling entirely. + * + * An internationalized domain may be given in either its unicode or its punycode form, and an IPv6 address + * has to be bracketed (`[::1]`). + */ + domains: string[]; + + /** + * Opens the per-domain queues, one per entry in `domains`, each under the alias `throttled-`. + * @default RequestQueue.open + */ + requestManagerOpener?: RequestManagerOpener; + + /** + * The delay applied after a domain's first HTTP 429, doubled on each subsequent one. + * @default 2 + */ + baseDelaySecs?: number; + + /** + * Upper bound on the delay between requests to a rate-limited domain, applied to both the exponential + * backoff and a `Retry-After` value. + * @default 60 + */ + maxDelaySecs?: number; + + /** + * How long a domain may rate-limit us without a single request getting through before the crawl is + * abandoned with a {@apilink PersistentRateLimitError}. + * + * A domain that keeps answering 429 for this long is not going to be crawled by waiting longer - the + * concurrency is too high for it, or it has blocked us outright. Its requests are deliberately left in + * their queue, so re-running the crawl without purging storages picks them up once the domain recovers. + * + * A crawler running with `keepAlive` is exempt - outliving a domain that will not let us through is the + * whole point there. + * @default 900 + */ + maxDomainStallSecs?: number; +} + +interface DomainState { + domain: string; + /** + * Earliest dispatch time imposed by 429 backoff, as a `Date.now()` timestamp. Kept apart from + * `crawlDelayUntil` so that a crawl-delay, which is armed on every single dispatch, cannot pass for an + * active backoff and swallow the 429s it is supposed to be tracking. + */ + backoffUntil: number; + /** Earliest dispatch time imposed by the robots.txt `Crawl-delay`, as a `Date.now()` timestamp. */ + crawlDelayUntil: number; + /** Time after which an incoming 429 is treated as a fresh burst rather than a continuation. */ + backoffDecaysAt: number; + consecutive429Count: number; + /** Minimum interval between dispatches, from a robots.txt `Crawl-delay` directive. */ + crawlDelayMs: number | null; + /** + * When the current unbroken run of 429s began, as a `Date.now()` timestamp, or `0` if the domain is not + * currently rate-limiting us. Cleared the moment a request gets through, so it measures how long we have + * been stonewalled rather than how long the domain has been quiet. + */ + rateLimitedSince: number; + /** + * When this domain last answered 429, as a `Date.now()` timestamp, or `0` if it never has. Read alongside + * `rateLimitedSince` to tell a domain that is still turning us away from one that is merely being waited out. + */ + lastRateLimitedAt: number; +} + +/** The moment a domain may be dispatched to again - whichever of its two independent clocks runs longer. */ +function throttledUntil(state: DomainState): number { + return Math.max(state.backoffUntil, state.crawlDelayUntil); +} + +/** + * A request manager that wraps another one and paces requests per domain. + * + * Requests for the configured {@apilink ThrottlingRequestManagerOptions.domains|`domains`} are routed into their own + * queue when they are added, so each request lives in exactly one place and deduplication keeps working. Everything + * else goes to the wrapped manager untouched. + * + * {@apilink ThrottlingRequestManager.fetchNextRequest|`fetchNextRequest()`} serves the domain that has been waiting + * longest and skips any that are backing off, falling back to the wrapped manager. It never blocks: while every + * remaining request belongs to a throttled domain it returns `null` and {@apilink ThrottlingRequestManager.isEmpty} + * reports `true`, so the crawler idles instead of holding a concurrency slot open. + * + * Delays come from two places: + * - HTTP 429 responses, honouring `Retry-After` and otherwise backing off exponentially. The crawlers report these + * automatically; a request that is throttled is retried later without counting against `maxRequestRetries` and + * without penalising its session. + * - robots.txt `Crawl-delay` directives, when `respectRobotsTxtFile` is enabled. + * + * This is opt-in: throttling only happens for a domain you list explicitly. + * + * **Example usage:** + * + * ```ts + * const crawler = new CheerioCrawler({ + * requestManager: new ThrottlingRequestManager({ + * inner: await RequestQueue.open(), + * domains: ['api.example.com', 'slow-site.org'], + * }), + * requestHandler: async ({ request }) => { ... }, + * }); + * ``` + * + * @category Sources + */ +export class ThrottlingRequestManager + implements IRequestManager, SupportsDomainThrottling +{ + private readonly inner: T; + private readonly requestManagerOpener: RequestManagerOpener; + private readonly baseDelayMs: number; + private readonly maxDelayMs: number; + private readonly maxDomainStallMs: number; + + private readonly domainStates = new Map(); + private readonly subManagers = new Map(); + private readonly log: CrawleeLogger; + + /** + * Sub-managers are keyed by a stable alias, so they outlive the process. They must therefore be reopened + * for every configured domain rather than created on first insert - otherwise a restart sees an empty map, + * reports the crawl finished, and strands whatever the previous run left in them. + */ + private subManagersReady?: Promise; + + /** Batches still being added in the background; keeps {@apilink ThrottlingRequestManager.isFinished} honest. */ + private inProgressBatchCount = 0; + + private readonly warnedAbout = new Set(); + + private get hasThrottledDomains(): boolean { + return this.domainStates.size > 0; + } + + constructor( + options: ThrottlingRequestManagerOptions, + private readonly config: Configuration = serviceLocator.getConfiguration(), + ) { + ow( + options, + ow.object.exactShape({ + inner: ow.object, + domains: ow.array.ofType(ow.string.nonEmpty), + requestManagerOpener: ow.optional.function, + baseDelaySecs: ow.optional.number.positive, + maxDelaySecs: ow.optional.number.positive, + maxDomainStallSecs: ow.optional.number.positive, + }), + ); + + this.inner = options.inner; + this.requestManagerOpener = + options.requestManagerOpener ?? + ((idOrAlias, opts) => RequestQueue.open(idOrAlias, opts) as unknown as Promise); + this.baseDelayMs = (options.baseDelaySecs ?? 2) * 1000; + this.maxDelayMs = (options.maxDelaySecs ?? 60) * 1000; + this.maxDomainStallMs = (options.maxDomainStallSecs ?? 900) * 1000; + this.log = serviceLocator.getLogger().child({ prefix: 'ThrottlingRequestManager' }); + + for (const domain of options.domains) { + let hostname: string; + try { + // These are bare hostnames, so they only reach `URL` - and with it IDNA - via a synthetic URL. + hostname = normalizeHostname(new URL(`http://${domain}`).hostname); + } catch { + throw new Error( + `"${domain}" is not a valid hostname. The \`domains\` option takes bare hostnames such as ` + + `"example.com"; an IPv6 address has to be bracketed, as in "[::1]".`, + ); + } + + this.domainStates.set(hostname, { + domain: hostname, + backoffUntil: 0, + crawlDelayUntil: 0, + backoffDecaysAt: 0, + consecutive429Count: 0, + crawlDelayMs: null, + rateLimitedSince: 0, + lastRateLimitedAt: 0, + }); + } + } + + /** The wrapped manager, holding every request whose domain is not throttled. */ + get innerManager(): T { + return this.inner; + } + + /** Warns once about sources that cannot be routed by domain, because their URLs are not known yet. */ + private warnIfNotRoutable(requestLike: Source): void { + if ('requestsFromUrl' in requestLike && requestLike.requestsFromUrl !== undefined && this.hasThrottledDomains) { + // The URL list is only fetched once the owning manager expands it, so we cannot know which domains + // it covers and cannot route it. Warn instead of silently exempting those URLs from throttling. + this.warnOnce( + 'urlListNotRouted', + `Requests loaded via \`requestsFromUrl\` cannot be routed to a per-domain queue, because their URLs ` + + `are not known at insertion time. They will be added to the inner request manager and will not ` + + `be throttled, even if they belong to a configured domain.`, + ); + } + } + + private warnOnce(key: string, message: string): void { + if (this.warnedAbout.has(key)) { + return; + } + this.warnedAbout.add(key); + this.log.warning(message); + } + + private extractDomain(url: string): string { + try { + return normalizeHostname(new URL(url).hostname); + } catch { + return ''; + } + } + + private getDomainState(url: string): DomainState | null { + const domain = this.extractDomain(url); + return this.domainStates.get(domain) ?? null; + } + + private async selectManager(url: string): Promise { + await this.ensureSubManagers(); + return this.managerForUrl(url); + } + + /** Only valid once {@apilink ThrottlingRequestManager.ensureSubManagers} has resolved. */ + private managerForUrl(url: string): T { + return this.subManagers.get(this.extractDomain(url)) ?? this.inner; + } + + private async ensureSubManagers(): Promise { + this.subManagersReady ??= (async () => { + await Promise.all( + Array.from(this.domainStates.keys(), async (domain) => { + const subManager = await this.requestManagerOpener( + // Backends use the alias as a directory name, and an IPv6 literal is full of characters + // Windows will not accept. Ordinary hostnames survive this untouched. + { alias: `throttled-${encodeURIComponent(domain)}` }, + { configuration: this.config }, + ); + this.subManagers.set(domain, subManager); + }), + ); + })(); + + await this.subManagersReady; + } + + private async getSubManagers(): Promise { + await this.ensureSubManagers(); + return Array.from(this.subManagers.values()); + } + + /** Configured domains that are not currently backing off, longest-overdue first. */ + private fetchableDomains(): string[] { + const now = Date.now(); + return Array.from(this.domainStates.values()) + .filter((state) => now >= throttledUntil(state)) + .sort((a, b) => throttledUntil(a) - throttledUntil(b)) + .map((state) => state.domain); + } + + /** + * Records a 429 response and puts the URL's domain into backoff. + * + * @returns `false` if the domain is not configured for throttling, in which case this is a no-op. + */ + recordDomainDelay(url: string, retryAfterMs?: number | null): boolean { + const state = this.getDomainState(url); + if (!state) { + return false; + } + + const now = Date.now(); + + // Recorded before the burst suppression below, because a suppressed 429 is still the domain turning us + // away - which is exactly what stall detection needs to know about. + state.lastRateLimitedAt = now; + if (state.rateLimitedSince === 0) { + state.rateLimitedSince = now; + } + + // Requests already in flight when the limit was hit all come back 429. They describe one rate-limit + // event, so only the first advances the backoff - otherwise concurrency alone drives the exponent. + // Only the backoff clock may suppress here: `crawlDelayUntil` is in the future after every dispatch, + // so consulting it would discard every 429 the domain ever sends, `Retry-After` included. + if (now < state.backoffUntil) { + return true; + } + + // A domain that has served us for a full extra backoff window is no longer rate-limiting; start over + // rather than carrying the old exponent into an unrelated burst. + if (now >= state.backoffDecaysAt) { + state.consecutive429Count = 0; + } + + state.consecutive429Count += 1; + + const retryAfterGiven = retryAfterMs !== undefined && retryAfterMs !== null; + let delayMs = retryAfterGiven ? retryAfterMs : this.baseDelayMs * Math.pow(2, state.consecutive429Count - 1); + + if (delayMs > this.maxDelayMs) { + const source = retryAfterGiven ? 'Retry-After header' : 'exponential backoff'; + this.log.warning( + `Capping ${source} delay of ${(delayMs / 1000).toFixed(1)}s for domain "${state.domain}" ` + + `to maxDelaySecs (${(this.maxDelayMs / 1000).toFixed(1)}s); the domain may continue to rate-limit. ` + + `Consider increasing maxDelaySecs if this recurs.`, + ); + delayMs = this.maxDelayMs; + } + + state.backoffUntil = now + delayMs; + state.backoffDecaysAt = state.backoffUntil + delayMs; + + this.log.info( + `Rate limit (429) detected for domain "${state.domain}" ` + + `(consecutive: ${state.consecutive429Count}, delay: ${(delayMs / 1000).toFixed(1)}s)`, + ); + + return true; + } + + /** + * Applies a robots.txt `Crawl-delay` to the URL's domain, as a minimum interval between dispatches. + * + * The first value wins, so a robots.txt re-fetch cannot change the cadence mid-crawl. + * + * @returns `false` if the domain is not configured for throttling, in which case this is a no-op. + */ + setCrawlDelay(url: string, delaySeconds: number): boolean { + const state = this.getDomainState(url); + if (!state) { + return false; + } + + if (state.crawlDelayMs === null) { + state.crawlDelayMs = delaySeconds * 1000; + this.log.debug(`Set crawl-delay for domain "${state.domain}" to ${delaySeconds}s`); + } + + return true; + } + + /** + * Throws {@apilink PersistentRateLimitError} if any domain has been rate-limiting us past + * {@apilink ThrottlingRequestManagerOptions.maxDomainStallSecs|`maxDomainStallSecs`} without letting a single + * request through. + * + * A domain qualifies only while it still has queued requests and is actively rate-limiting - a domain that + * has simply run out of work is finished, not stalled, and one being waited out under a long robots.txt + * `Crawl-delay` is being obeyed, not stonewalled. + */ + async assertNoStalledDomains(): Promise { + await this.ensureSubManagers(); + + const now = Date.now(); + const candidates = Array.from(this.domainStates.values()).filter( + // Together: it is still turning us away, and has been doing so without a break for longer than the + // window. A domain that has simply been idle starts this clock at its first 429 rather than + // arriving with the idle time already on it. + (state) => + state.rateLimitedSince !== 0 && + now - state.lastRateLimitedAt <= this.maxDomainStallMs && + now - state.rateLimitedSince > this.maxDomainStallMs, + ); + + const stalled = ( + await Promise.all( + candidates.map(async (state) => ((await this.subManagers.get(state.domain)!.isEmpty()) ? null : state)), + ) + ).filter((state) => state !== null); + + if (stalled.length === 0) { + return; + } + + const summary = stalled + .map((state) => `"${state.domain}" (${((now - state.rateLimitedSince) / 1000).toFixed(0)}s)`) + .join(', '); + + throw new PersistentRateLimitError( + `Giving up: ${summary} rate-limited every request for longer than maxDomainStallSecs ` + + `(${(this.maxDomainStallMs / 1000).toFixed(0)}s). Waiting longer will not help - lower the ` + + `crawler's concurrency, or drop these domains. Their requests are still queued, so re-running ` + + `without purging storages will resume them if the rate limit lifts.`, + ); + } + + /** Records that a domain let a request through, which ends any rate-limit run stall detection was timing. */ + private recordProgress(url: string): void { + const state = this.getDomainState(url); + if (state) { + state.rateLimitedSince = 0; + } + } + + // --- IRequestManager Implementation --- + + async addRequest(requestLike: Source, options?: RequestQueueOperationOptions): Promise { + this.warnIfNotRoutable(requestLike); + + const manager = await this.selectManager(requestLike.url ?? ''); + return manager.addRequest(requestLike, options); + } + + /** + * Adds requests in batches, routing each one to the manager that owns its domain. + * + * Batching, validation, deduplication and `Retry-After`-free bookkeeping are all delegated to the target + * managers - this only decides where each request goes, one batch at a time, so a lazy or unbounded input + * iterable is never fully materialized. + */ + async addRequestsBatched( + requests: RequestsLike, + options: AddRequestsBatchedOptions = {}, + ): Promise { + await this.ensureSubManagers(); + + // Normalized up front so the shared batching helper - and `requestsOverLimit` - only ever see `Source`. + async function* iterateRequests(): AsyncGenerator { + for await (const request of asyncifyIterable(requests)) { + yield typeof request === 'string' ? { url: request } : request; + } + } + + return drainRequestBatches({ + items: iterateRequests(), + batchSize: options.batchSize ?? 1000, + waitBetweenBatchesMillis: options.waitBetweenBatchesMillis ?? 1000, + waitForAllRequestsToBeAdded: options.waitForAllRequestsToBeAdded ?? false, + maxNewRequests: options.maxNewRequests, + + // Routing is the only thing this manager adds; the targets do the batching, validation and + // deduplication themselves. + processChunk: async (chunk) => { + const byManager = new Map(); + for (const request of chunk) { + this.warnIfNotRoutable(request); + + const manager = this.managerForUrl(request.url ?? ''); + const bucket = byManager.get(manager); + if (bucket) { + bucket.push(request); + } else { + byManager.set(manager, [request]); + } + } + + const results = await Promise.all( + Array.from(byManager, ([manager, slice]) => + manager.addRequestsBatched(slice, { + forefront: options.forefront, + // The slice is already one batch, and we need its results before releasing the next one. + batchSize: slice.length, + waitForAllRequestsToBeAdded: true, + }), + ), + ); + + return results.flatMap((result) => result.addedRequests); + }, + + // Keeps the crawler from concluding it is finished while batches are still landing. + trackBackgroundBatches: (batches) => { + this.inProgressBatchCount += 1; + void batches.finally(() => { + this.inProgressBatchCount -= 1; + }); + }, + }); + } + + async reclaimRequest( + request: Request, + options?: RequestQueueOperationOptions, + ): Promise { + const manager = await this.selectManager(request.url); + return manager.reclaimRequest(request, options); + } + + async markRequestAsHandled(request: Request): Promise { + const manager = await this.selectManager(request.url); + // Reached whether the request succeeded or ran out of retries; either way the domain answered us. + this.recordProgress(request.url); + return manager.markRequestAsHandled(request); + } + + async getTotalCount(): Promise { + return this.sumOverManagers((manager) => manager.getTotalCount()); + } + + async getPendingCount(): Promise { + return this.sumOverManagers((manager) => manager.getPendingCount()); + } + + async getHandledCount(): Promise { + return this.sumOverManagers((manager) => manager.getHandledCount()); + } + + /** + * Whether the next {@apilink ThrottlingRequestManager.fetchNextRequest} would return `null`. + * + * Requests waiting on a throttled domain count as unavailable, so a crawler whose task loop is gated on + * this idles for the backoff instead of spinning on a fetch that cannot succeed yet. + */ + async isEmpty(): Promise { + await this.ensureSubManagers(); + + const fetchable = [this.inner, ...this.fetchableDomains().map((domain) => this.subManagers.get(domain)!)]; + const results = await Promise.all(fetchable.map((manager) => manager.isEmpty())); + + return results.every(Boolean); + } + + /** Unlike {@apilink ThrottlingRequestManager.isEmpty}, throttled requests still count as outstanding work. */ + async isFinished(): Promise { + if (this.inProgressBatchCount > 0) { + return false; + } + + return this.everyManager((manager) => manager.isFinished()); + } + + /** + * Empties every manager and clears the accumulated backoff. A robots.txt `Crawl-delay` is a property of the + * site rather than of the run, so it survives. + */ + async purge(): Promise { + await this.forEachManager((manager) => manager.purge?.()); + for (const state of this.domainStates.values()) { + state.consecutive429Count = 0; + state.backoffUntil = 0; + state.crawlDelayUntil = 0; + state.backoffDecaysAt = 0; + state.rateLimitedSince = 0; + state.lastRateLimitedAt = 0; + } + } + + async setExpectedRequestProcessingTimeSecs(secs: number): Promise { + await this.forEachManager((manager) => manager.setExpectedRequestProcessingTimeSecs?.(secs)); + } + + private async forEachManager(fn: (manager: T) => Promise | undefined): Promise { + // `fn` targets optional members, so it may return nothing - the wrapper normalizes that for `Promise.all`. + await Promise.all([this.inner, ...(await this.getSubManagers())].map(async (manager) => fn(manager))); + } + + private async sumOverManagers(fn: (manager: T) => Promise): Promise { + const counts = await Promise.all([this.inner, ...(await this.getSubManagers())].map(fn)); + return counts.reduce((a, b) => a + b, 0); + } + + private async everyManager(fn: (manager: T) => Promise): Promise { + const results = await Promise.all([this.inner, ...(await this.getSubManagers())].map(fn)); + return results.every(Boolean); + } + + /** + * Returns the next request from a domain that is not backing off, or from the inner manager. + * + * Returns `null` while every remaining request belongs to a throttled domain - it never waits the backoff + * out, because a consumer parked in here holds a concurrency slot, which the autoscaler reads as spare + * capacity and answers by scaling up. Callers poll instead, and {@apilink ThrottlingRequestManager.isEmpty} + * reports `true` meanwhile so the crawler's task loop idles rather than spins. + */ + async fetchNextRequest(): Promise | null> { + await this.ensureSubManagers(); + + for (const domain of this.fetchableDomains()) { + const state = this.domainStates.get(domain)!; + + // Armed while the fetch below is still suspended, so that a concurrent `fetchNextRequest` cannot + // find the domain fetchable and dispatch into the same window - which would pace each task + // rather than the domain. + const crawlDelayBefore = state.crawlDelayUntil; + if (state.crawlDelayMs !== null) { + state.crawlDelayUntil = Date.now() + state.crawlDelayMs; + } + + const request = await this.subManagers.get(domain)!.fetchNextRequest(); + if (request) { + return request; + } + + // No dispatch to pace, so the domain keeps its slot. + state.crawlDelayUntil = crawlDelayBefore; + } + + return this.inner.fetchNextRequest(); + } + + async *[Symbol.asyncIterator]() { + while (true) { + const req = await this.fetchNextRequest(); + if (!req) break; + yield req; + } + } + + async persistState(): Promise { + await this.forEachManager((manager) => manager.persistState?.()); + } + + async drop(): Promise { + await this.forEachManager((manager) => (manager as { drop?(): Promise }).drop?.()); + this.subManagers.clear(); + this.subManagersReady = undefined; + } +} diff --git a/packages/core/src/url.ts b/packages/core/src/url.ts new file mode 100644 index 000000000000..077c0d5a2bd1 --- /dev/null +++ b/packages/core/src/url.ts @@ -0,0 +1,11 @@ +/** + * The canonical form of a hostname: lower-case, punycode, and without the optional root dot. + * + * Pass both sides of a hostname comparison through this, so that a domain written as `háčky.cz` still matches the + * `xn--hky-ela4t.cz` that `URL` reports. + * + * @internal + */ +export function normalizeHostname(hostname: string): string { + return hostname.toLowerCase().replace(/\.$/, ''); +} diff --git a/packages/http-crawler/src/internals/http-crawler.ts b/packages/http-crawler/src/internals/http-crawler.ts index a118efae115e..e40ac25de9db 100644 --- a/packages/http-crawler/src/internals/http-crawler.ts +++ b/packages/http-crawler/src/internals/http-crawler.ts @@ -26,7 +26,7 @@ import { Router, SessionError, } from '@crawlee/basic'; -import { type LoadedRequest, getCookiesFromResponse } from '@crawlee/core'; +import { type LoadedRequest, RequestThrottledError, getCookiesFromResponse } from '@crawlee/core'; import { ResponseWithUrl } from '@crawlee/http-client'; import type { Awaitable, Dictionary, ISession } from '@crawlee/types'; import { type CheerioRoot, RETRY_CSS_SELECTORS } from '@crawlee/utils/internal'; @@ -572,6 +572,18 @@ export class HttpCrawler< tryCancel(); + // Before `parseResponse`, which throws for error status codes - a 429 the user opted into treating as an + // error is still a rate limit the domain should back off from. + if (crawlingContext.response.status === 429) { + const retryAfter = crawlingContext.response.headers.get('retry-after'); + if (this.recordDomainRateLimit(crawlingContext.request.url, retryAfter)) { + // This is the one path that never reads the body, so cancel it to release the connection + // rather than leaving it to the garbage collector. + await crawlingContext.response.body?.cancel().catch(() => {}); + throw new RequestThrottledError(`${crawlingContext.request.url} responded with 429.`); + } + } + // Reading the body is still part of the navigation, so it draws from the same shared window: on a server // that streams the body slowly the request completes (headers arrive) but the body read would otherwise // run unbounded. `extendTimeout` from a post-navigation hook has already pushed this deadline out if asked. diff --git a/packages/utils/src/internals/robots.ts b/packages/utils/src/internals/robots.ts index 614a9b98fac2..71ef9ac31d3c 100644 --- a/packages/utils/src/internals/robots.ts +++ b/packages/utils/src/internals/robots.ts @@ -24,11 +24,15 @@ import { Sitemap } from './sitemap.js'; * ``` */ export class RobotsTxtFile { - #robots: Pick; + #robots: Pick; #proxyUrl?: string; #logger?: CrawleeLogger; - private constructor(robots: Pick, proxyUrl?: string, logger?: CrawleeLogger) { + private constructor( + robots: Pick, + proxyUrl?: string, + logger?: CrawleeLogger, + ) { this.#robots = robots; this.#proxyUrl = proxyUrl; this.#logger = logger; @@ -101,6 +105,9 @@ export class RobotsTxtFile { getSitemaps() { return []; }, + getCrawlDelay() { + return undefined; + }, }, proxyUrl, logger, @@ -111,6 +118,14 @@ export class RobotsTxtFile { return new RobotsTxtFile(robotsParser(url.toString(), await response.text()), proxyUrl, logger); } + /** + * Get crawl delay for a given user agent. + * @param [userAgent] relevant user agent, default to `*` + */ + getCrawlDelay(userAgent = '*'): number | undefined { + return this.#robots.getCrawlDelay(userAgent); + } + /** * Check if a URL should be crawled by robots. * @param url the URL to check against the rules in robots.txt diff --git a/test/core/crawlers/basic_crawler.test.ts b/test/core/crawlers/basic_crawler.test.ts index 9cfef6719119..99e51377609c 100644 --- a/test/core/crawlers/basic_crawler.test.ts +++ b/test/core/crawlers/basic_crawler.test.ts @@ -3,7 +3,14 @@ import type { Server } from 'node:http'; import http from 'node:http'; import type { AddressInfo } from 'node:net'; -import type { EnqueueLinksOptions, ErrorHandler, RequestHandler, RequestOptions, Source } from '@crawlee/basic'; +import type { + BasicCrawlerOptions, + EnqueueLinksOptions, + ErrorHandler, + RequestHandler, + RequestOptions, + Source, +} from '@crawlee/basic'; import type { Session } from '@crawlee/basic'; import { BasicCrawler, @@ -25,6 +32,7 @@ import { serviceLocator, SessionPool, Statistics, + ThrottlingRequestManager, } from '@crawlee/basic'; import type { CalculatedStatistics, IConcurrencySystem, IStatistics } from '@crawlee/core'; import { ConcurrencySystem, MemoryStorageBackend, RequestState } from '@crawlee/core'; @@ -177,6 +185,23 @@ describe('BasicCrawler', () => { expect(await requestList.isEmpty()).toBe(true); }); + test('accepts a `requestManager` and crawls from it', async () => { + const requestManager = await RequestQueue.open(); + await requestManager.addRequest({ url: 'https://example.com/from-request-manager' }); + + const processed: string[] = []; + const crawler = new BasicCrawler({ + requestManager, + requestHandler: async ({ request }) => { + processed.push(request.url); + }, + }); + + await crawler.run(); + + expect(processed).toEqual(['https://example.com/from-request-manager']); + }); + test('folds a supplied concurrencySystem into its pool and never tears the system down', async () => { const sources = [...Array(20).keys()].map((index) => ({ url: `https://example.com/${index}` })); const requestList = await RequestList.open(null, sources); @@ -2435,6 +2460,61 @@ describe('BasicCrawler', () => { expect(addRequestsBatchedSpy).toHaveBeenCalledOnce(); }); + describe('robots.txt crawl-delay', () => { + const crawlerWithCrawlDelay = (options: Partial) => + new (class MockedRobotsTxtCrawler extends BasicCrawler { + override async getRobotsTxtFileForUrl(_: string) { + return RobotsTxtFile.from('http://example.com/robots.txt', 'User-agent: *\nCrawl-delay: 5\n'); + } + })({ respectRobotsTxtFile: true, requestHandler: async () => {}, ...options } as BasicCrawlerOptions); + + test('is applied when a ThrottlingRequestManager covers the domain', async () => { + const requestManager = new ThrottlingRequestManager({ + inner: await RequestQueue.open(), + domains: ['example.com'], + }); + const crawler = crawlerWithCrawlDelay({ requestManager }); + const warning = vitest.spyOn(crawler.log, 'warning').mockImplementation(() => {}); + + await crawler.addRequests(['http://example.com/1', 'http://example.com/2']); + + expect(warning).not.toHaveBeenCalled(); + + // The robots.txt `Crawl-delay: 5` must have reached the manager, not merely been survivable. + await requestManager.fetchNextRequest(); + const state = (requestManager as any).domainStates.get('example.com'); + expect(state.crawlDelayMs).toBe(5_000); + + // ...and it paces dispatch: the next request is held back rather than served immediately. + expect(state.crawlDelayUntil).toBeGreaterThan(Date.now() + 4_000); + expect(await requestManager.fetchNextRequest()).toBeNull(); + }); + + test('warns when the request manager cannot honour it', async () => { + const crawler = crawlerWithCrawlDelay({ requestQueue: await RequestQueue.open() }); + const warning = vitest.spyOn(crawler.log, 'warning').mockImplementation(() => {}); + + await crawler.addRequests(['http://example.com/1', 'http://example.com/2']); + + expect(warning).toHaveBeenCalledTimes(1); + expect(warning.mock.calls[0][0]).toMatch(/crawl-delay of 5s/); + }); + + test('warns when the domain is missing from the manager `domains` list', async () => { + const requestManager = new ThrottlingRequestManager({ + inner: await RequestQueue.open(), + domains: ['some-other-domain.com'], + }); + const crawler = crawlerWithCrawlDelay({ requestManager }); + const warning = vitest.spyOn(crawler.log, 'warning').mockImplementation(() => {}); + + await crawler.addRequests(['http://example.com/1']); + + expect(warning).toHaveBeenCalledTimes(1); + expect(warning.mock.calls[0][0]).toMatch(/example\.com/); + }); + }); + test('enqueueLinks should respect custom user-agent robots.txt rules', async () => { const requestQueue = await RequestQueue.open(); const visitedUrls: string[] = []; @@ -2491,7 +2571,7 @@ describe('BasicCrawler', () => { const crawler = new (class MockedRobotsTxtCrawler extends BasicCrawler { override async getRobotsTxtFileForUrl(_: string) { - return { isAllowed: isAllowedSpy } as unknown as RobotsTxtFile; + return { isAllowed: isAllowedSpy, getCrawlDelay: () => undefined } as unknown as RobotsTxtFile; } })({ requestQueue, diff --git a/test/core/crawlers/http_crawler.test.ts b/test/core/crawlers/http_crawler.test.ts index 351143b04afb..15da3b3ae3bf 100644 --- a/test/core/crawlers/http_crawler.test.ts +++ b/test/core/crawlers/http_crawler.test.ts @@ -4,8 +4,16 @@ import { Readable } from 'node:stream'; import type { ConcurrencySystemOptions } from '@crawlee/core'; import { MemoryStorageBackend, serviceLocator } from '@crawlee/core'; -import { ConcurrencySystem, HttpCrawler, SessionPool } from '@crawlee/http'; +import { + ConcurrencySystem, + HttpCrawler, + PersistentRateLimitError, + RequestQueue, + SessionPool, + ThrottlingRequestManager, +} from '@crawlee/http'; import { ResponseWithUrl } from '@crawlee/http-client'; +import { sleep } from '@crawlee/utils'; import iconv from 'iconv-lite'; const router = new Map(); @@ -562,3 +570,122 @@ test('works with a custom HttpClient', async () => { expect(results[0].includes('Schmexample Domain')).toBeTruthy(); expect(results[1].includes('Schmexample Domain')).toBeTruthy(); }); + +test('a 429 on a throttled domain paces the retry without spending it or the session', async () => { + const hits: number[] = []; + router.set('/429-then-ok', (req, res) => { + hits.push(Date.now()); + if (hits.length === 1) { + res.statusCode = 429; + res.setHeader('retry-after', '1'); + res.end(); + return; + } + res.setHeader('content-type', 'text/html'); + res.end('ok'); + }); + + const throttlingManager = new ThrottlingRequestManager({ + inner: await RequestQueue.open(), + domains: ['127.0.0.1'], + }); + + const sessionPool = new SessionPool(); + const retiredSessions: string[] = []; + const markedBad: string[] = []; + + const handled: string[] = []; + const crawler = new HttpCrawler({ + requestManager: throttlingManager, + sessionPool, + maxRequestRetries: 0, + preNavigationHooks: [ + async ({ session }) => { + vitest.spyOn(session!, 'retire').mockImplementation(() => retiredSessions.push(session!.id)); + vitest.spyOn(session!, 'markBad').mockImplementation(() => markedBad.push(session!.id)); + }, + ], + requestHandler: async ({ request }) => { + handled.push(request.url); + }, + }); + + const stats = await crawler.run([`${url}/429-then-ok`]); + + // `maxRequestRetries: 0` would have failed the request outright had the 429 been charged as a retry. + expect(handled).toHaveLength(1); + expect(stats.requestsFailed).toBe(0); + + // The domain's `Retry-After` was honoured between the two attempts... + expect(hits).toHaveLength(2); + expect(hits[1] - hits[0]).toBeGreaterThanOrEqual(1000); + + // ...and the session came out untouched - a rate limit says nothing about it. + expect(retiredSessions).toEqual([]); + expect(markedBad).toEqual([]); +}, 30_000); + +test('a domain that never stops rate-limiting shuts the crawl down instead of hanging', async () => { + let hits = 0; + router.set('/always-429', (req, res) => { + hits++; + res.statusCode = 429; + res.end(); + }); + + const crawler = new HttpCrawler({ + requestManager: new ThrottlingRequestManager({ + inner: await RequestQueue.open(), + domains: ['127.0.0.1'], + baseDelaySecs: 0.05, + maxDelaySecs: 0.1, + maxDomainStallSecs: 2, + }), + maxRequestRetries: 0, + requestHandler: async () => {}, + }); + + // Without the stall detector this never resolves - a throttled request costs no retries. + await expect(crawler.run([`${url}/always-429`])).rejects.toThrow(PersistentRateLimitError); + + expect(hits).toBeGreaterThan(1); + + // The request is deliberately left queued, so a later run can pick it up if the rate limit lifts. + expect(await crawler.getRequestManager().then((manager) => manager.getPendingCount())).toBe(1); +}, 30_000); + +test('`keepAlive` outlives a domain that never stops rate-limiting', async () => { + router.set('/always-429-keep-alive', (req, res) => { + res.statusCode = 429; + res.end(); + }); + + const crawler = new HttpCrawler({ + requestManager: new ThrottlingRequestManager({ + inner: await RequestQueue.open(), + domains: ['127.0.0.1'], + baseDelaySecs: 0.05, + maxDelaySecs: 0.1, + maxDomainStallSecs: 0.5, + }), + keepAlive: true, + maxRequestRetries: 0, + requestHandler: async () => {}, + }); + + const running = crawler.run([`${url}/always-429-keep-alive`]); + + // Several times the stall threshold - long enough that the shutdown would have fired by now. + const outcome = await Promise.race([ + running.then( + () => 'shut down', + () => 'threw', + ), + sleep(3000).then(() => 'still running' as const), + ]); + + expect(outcome).toBe('still running'); + + await crawler.teardown(); + await running; +}, 30_000); diff --git a/test/core/http.test.ts b/test/core/http.test.ts new file mode 100644 index 000000000000..81851216facc --- /dev/null +++ b/test/core/http.test.ts @@ -0,0 +1,29 @@ +import { parseRetryAfterHeader } from '@crawlee/core'; + +test('parseRetryAfterHeader parses delay-seconds and HTTP-dates', () => { + expect(parseRetryAfterHeader('120')).toBe(120_000); + expect(parseRetryAfterHeader(' 5 ')).toBe(5000); + // Zero-padded values are valid `delay-seconds`. + expect(parseRetryAfterHeader('05')).toBe(5000); + + // A zero delay names no deadline; reporting it as one would leave the domain unthrottled and busy-loop. + expect(parseRetryAfterHeader('0')).toBeNull(); + expect(parseRetryAfterHeader('00')).toBeNull(); + + // date format + const futureDate = new Date(Date.now() + 5000).toUTCString(); + const delay = parseRetryAfterHeader(futureDate); + expect(delay).toBeGreaterThan(0); + expect(delay).toBeLessThanOrEqual(5500); + + // A date in the past means "no delay", not a negative one. + expect(parseRetryAfterHeader(new Date(Date.now() - 5000).toUTCString())).toBeNull(); + + expect(parseRetryAfterHeader(null)).toBeNull(); + expect(parseRetryAfterHeader(undefined)).toBeNull(); + expect(parseRetryAfterHeader('')).toBeNull(); + expect(parseRetryAfterHeader('invalid')).toBeNull(); + // Not `delay-seconds`; a negative delay would have suppressed the backoff entirely. + expect(parseRetryAfterHeader('-5')).toBeNull(); + expect(parseRetryAfterHeader('1.5')).toBeNull(); +}); diff --git a/test/core/storages/request_queue.test.ts b/test/core/storages/request_queue.test.ts index afdfa4ac901a..36977f30d9a5 100644 --- a/test/core/storages/request_queue.test.ts +++ b/test/core/storages/request_queue.test.ts @@ -654,3 +654,32 @@ describe('RequestQueue (request lifecycle)', () => { expect(retrievedUrls.map((x) => new URL(x).pathname)).toEqual(Array.from({ length: 5 }, (_, i) => `/${i + 1}`)); }); }); + +describe('RequestQueue background batches', () => { + beforeEach(async () => { + serviceLocator.setStorageBackend(new MemoryStorageBackend()); + }); + + test('a failing background batch rejects instead of hanging, and stops blocking isFinished', async () => { + const queue = await RequestQueue.open(); + + let batches = 0; + const original = queue.addRequests.bind(queue); + vitest.spyOn(queue, 'addRequests').mockImplementation(async (requests, options) => { + if (++batches > 1) throw new Error('backend exploded'); + return original(requests, options); + }); + + const result = await queue.addRequestsBatched( + [{ url: 'https://example.com/1' }, { url: 'https://example.com/2' }], + { batchSize: 1, waitBetweenBatchesMillis: 0 }, + ); + + // Previously the async promise executor swallowed the throw: this promise never settled at all. + await expect(result.waitForAllRequestsToBeAdded).rejects.toThrow('backend exploded'); + + // ...and the in-flight batch counter stayed stuck, so the queue claimed to be unfinished forever. + await sleep(10); + expect(queue['inProgressRequestBatchCount']).toBe(0); + }, 10_000); +}); diff --git a/test/core/storages/throttling_request_manager.test.ts b/test/core/storages/throttling_request_manager.test.ts new file mode 100644 index 000000000000..4b542ae2ca32 --- /dev/null +++ b/test/core/storages/throttling_request_manager.test.ts @@ -0,0 +1,543 @@ +import type { AddRequestsBatchedResult } from '@crawlee/core'; +import { + MemoryStorageBackend, + PersistentRateLimitError, + RequestQueue, + serviceLocator, + ThrottlingRequestManager, + withStorageTransaction, +} from '@crawlee/core'; +import { sleep } from '@crawlee/utils'; + +describe('ThrottlingRequestManager', () => { + beforeEach(() => { + serviceLocator.setStorageBackend(new MemoryStorageBackend()); + }); + + async function createQueue(name = 'inner-queue') { + return RequestQueue.open({ name }); + } + + function domainState(manager: ThrottlingRequestManager, domain: string) { + return (manager as any).domainStates.get(domain); + } + + /** Models the crawler's task loop: poll, and idle while the manager reports itself empty. */ + async function pollForNextRequest(manager: ThrottlingRequestManager, timeoutMs = 5000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const request = await manager.fetchNextRequest(); + if (request) { + return request; + } + await sleep(20); + } + throw new Error('Timed out waiting for a request to become available'); + } + + test('Routing: requests to configured domains route to sub-managers, others to inner queue', async () => { + const inner = await createQueue(); + const manager = new ThrottlingRequestManager({ + inner, + domains: ['example.com'], + }); + + // Add request to inner domain + await manager.addRequest({ url: 'https://other.com/a' }); + // Add request to throttled domain + await manager.addRequest({ url: 'https://example.com/a' }); + + expect(await inner.getTotalCount()).toBe(1); + expect(await manager.getTotalCount()).toBe(2); + + // Fetching next request should yield them + const req1 = await manager.fetchNextRequest(); + expect(req1!.url).toBe('https://example.com/a'); // Throttled domains checked first + + const req2 = await manager.fetchNextRequest(); + expect(req2!.url).toBe('https://other.com/a'); + + expect(await manager.fetchNextRequest()).toBeNull(); + }); + + test('Routing: a configured domain matches however its hostname is spelled', async () => { + const inner = await createQueue(); + const manager = new ThrottlingRequestManager({ + inner, + // A unicode, a punycode and a root-dotted spelling. + domains: ['HÁČKY.cz', 'xn--bcher-kva.example', 'example.com.', '[::1]'], + }); + + await manager.addRequest({ url: 'https://xn--hky-ela4t.cz/punycode' }); + await manager.addRequest({ url: 'https://háčky.cz./unicode-with-root-dot' }); + await manager.addRequest({ url: 'https://bücher.example/unicode' }); + await manager.addRequest({ url: 'https://example.com/no-root-dot' }); + await manager.addRequest({ url: 'http://[::1]:8080/bracketed' }); + + // Every one of them belongs to a configured domain, so none of them fell through to the inner queue. + expect(await inner.getTotalCount()).toBe(0); + expect(await manager.getTotalCount()).toBe(5); + }); + + test('rejects a domain that is not a hostname', async () => { + const inner = await createQueue(); + + // Unbracketed IPv6 - previously accepted, then silently matched nothing. + expect(() => new ThrottlingRequestManager({ inner, domains: ['::1'] })).toThrow(/not a valid hostname/); + }); + + test('addRequestsBatched routing', async () => { + const inner = await createQueue(); + const manager = new ThrottlingRequestManager({ + inner, + domains: ['example.com', 'foo.com'], + }); + + await manager.addRequestsBatched([ + { url: 'https://example.com/1' }, + { url: 'https://other.com/1' }, + { url: 'https://foo.com/1' }, + ]); + + expect(await inner.getTotalCount()).toBe(1); + expect(await manager.getTotalCount()).toBe(3); + }); + + test('addRequestsBatched consumes the input lazily and honours maxNewRequests', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + }); + + let produced = 0; + async function* urls() { + for (let i = 0; i < 10; i++) { + produced++; + yield { url: `https://example.com/${i}` }; + } + } + + const result = await manager.addRequestsBatched(urls(), { batchSize: 2, maxNewRequests: 4 }); + + expect(await manager.getTotalCount()).toBe(4); + expect(result.requestsOverLimit).toHaveLength(6); + // Everything was pulled to report the leftovers, but only via the budget-capped chunks. + expect(produced).toBe(10); + }); + + test('addRequestsBatched keeps isFinished false while background batches are landing', async () => { + const inner = await createQueue(); + // Reports itself done the moment each batch lands, so only our own batch bookkeeping can hold the crawl open. + const eagerlyFinished = { + ...inner, + addRequestsBatched: inner.addRequestsBatched.bind(inner), + getTotalCount: inner.getTotalCount.bind(inner), + isEmpty: async () => true, + isFinished: async () => true, + } as unknown as RequestQueue; + + const manager = new ThrottlingRequestManager({ inner: eagerlyFinished, domains: [] }); + + const result = await manager.addRequestsBatched( + [{ url: 'https://other.com/1' }, { url: 'https://other.com/2' }], + { batchSize: 1, waitBetweenBatchesMillis: 0 }, + ); + + // The inner manager reports itself finished as soon as its own batch lands, but ours must not - + // there is still a batch in flight behind it. + expect(await manager.isFinished()).toBe(false); + + await result.waitForAllRequestsToBeAdded; + expect(await manager.getTotalCount()).toBe(2); + }); + + test('addRequestsBatched surfaces a background failure without an unhandled rejection', async () => { + const inner = await createQueue(); + let batches = 0; + const flaky = { + ...inner, + addRequestsBatched: async (...args: Parameters) => { + if (++batches > 1) { + throw new Error('backend exploded'); + } + return inner.addRequestsBatched(...args); + }, + } as unknown as RequestQueue; + + const manager = new ThrottlingRequestManager({ inner: flaky, domains: [] }); + + const result = await manager.addRequestsBatched( + [{ url: 'https://other.com/1' }, { url: 'https://other.com/2' }], + { batchSize: 1, waitBetweenBatchesMillis: 0 }, + ); + + // Nobody is obliged to await this - but if it rejects unhandled, Node kills the process. + await expect(result.waitForAllRequestsToBeAdded).rejects.toThrow('backend exploded'); + }); + + describe('addRequestsBatched in a transaction', () => { + const sixRequests = Array.from({ length: 6 }, (_, i) => ({ url: `https://example.com/${i}` })); + + test('deferred: every batch is rolled back, not just the first', async () => { + const manager = new ThrottlingRequestManager({ inner: await createQueue(), domains: ['example.com'] }); + + let result!: AddRequestsBatchedResult; + await withStorageTransaction( + async (transaction) => { + result = await manager.addRequestsBatched(sixRequests, { + batchSize: 2, + waitBetweenBatchesMillis: 0, + }); + transaction.rollback(); + }, + { policy: { requestQueue: 'deferred' } }, + ); + await result.waitForAllRequestsToBeAdded; + + expect(await manager.getTotalCount()).toBe(0); + }); + + test('write-through: batches that outlive the transaction stay out of its journal', async () => { + const manager = new ThrottlingRequestManager({ inner: await createQueue(), domains: ['example.com'] }); + + await withStorageTransaction(async (transaction) => { + const result = await manager.addRequestsBatched(sixRequests, { + batchSize: 2, + waitBetweenBatchesMillis: 0, + }); + await result.waitForAllRequestsToBeAdded; + + // Only the initial batch was added within the transaction's scope; the rest writes directly, + // exactly as a bare `RequestQueue` does. + expect(transaction.enqueuedUrls).toHaveLength(2); + }); + }); + }); + + test('warns that requestsFromUrl sources cannot be domain-routed', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + }); + const warning = vitest.spyOn((manager as any).log, 'warning').mockImplementation(() => {}); + + await manager.addRequestsBatched([ + { requestsFromUrl: 'https://example.com/urls.txt' }, + { requestsFromUrl: 'https://example.com/more.txt' }, + ]); + + expect(warning).toHaveBeenCalledTimes(1); + expect(warning.mock.calls[0][0]).toMatch(/requestsFromUrl/); + }); + + test('recordDomainDelay enforces throttling and fair scheduling', async () => { + const inner = await createQueue(); + const manager = new ThrottlingRequestManager({ + inner, + domains: ['example.com', 'foo.com'], + baseDelaySecs: 0.1, + }); + + await manager.addRequest({ url: 'https://example.com/1' }); + await manager.addRequest({ url: 'https://foo.com/1' }); + + // Record a 500ms delay on example.com + const recorded = manager.recordDomainDelay('https://example.com/1', 500); + expect(recorded).toBe(true); + + // Fetch next request - should fetch foo.com since example.com is throttled + const req1 = await manager.fetchNextRequest(); + expect(req1!.url).toBe('https://foo.com/1'); + + // example.com is still throttled and inner is empty, so there is nothing to fetch right now - + // and the manager must say so rather than block the caller. + expect(await manager.fetchNextRequest()).toBeNull(); + expect(await manager.isEmpty()).toBe(true); + // ...while still reporting the throttled request as outstanding work. + expect(await manager.isFinished()).toBe(false); + + const start = Date.now(); + const req2 = await pollForNextRequest(manager); + + expect(Date.now() - start).toBeGreaterThanOrEqual(400); + expect(req2.url).toBe('https://example.com/1'); + }); + + test('a burst of concurrent 429s advances the backoff only once', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + baseDelaySecs: 0.05, + maxDelaySecs: 60, + }); + + // Eight requests were already in flight when the limit was hit; they all come back 429. + for (let i = 0; i < 8; i++) { + expect(manager.recordDomainDelay('https://example.com/1')).toBe(true); + } + + expect(domainState(manager, 'example.com').consecutive429Count).toBe(1); + }); + + test('the backoff decays once the domain has stopped rate-limiting', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + baseDelaySecs: 10, + maxDelaySecs: 60, + }); + const state = domainState(manager, 'example.com'); + + manager.recordDomainDelay('https://example.com/1'); + + // Rewinding both clocks beats sleeping out real delays - a loaded CI box cannot race it. + const rewind = (ms: number) => { + state.backoffUntil -= ms; + state.backoffDecaysAt -= ms; + }; + + // Past the backoff but still inside the decay window: the next 429 continues the same burst. + rewind(11_000); + manager.recordDomainDelay('https://example.com/1'); + expect(state.consecutive429Count).toBe(2); + + // Past the decay window as well: the domain is treated as recovered and the exponent restarts. + rewind(41_000); + manager.recordDomainDelay('https://example.com/1'); + expect(state.consecutive429Count).toBe(1); + }); + + test('caps the delay at maxDelaySecs', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + maxDelaySecs: 1, + }); + + manager.recordDomainDelay('https://example.com/1', 3_600_000); + + expect(domainState(manager, 'example.com').backoffUntil).toBeLessThanOrEqual(Date.now() + 1000); + }); + + test('fetchNextRequest does not block while a domain is throttled', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + maxDelaySecs: 60, + }); + + await manager.addRequest({ url: 'https://example.com/1' }); + manager.recordDomainDelay('https://example.com/1', 60_000); + + const start = Date.now(); + expect(await manager.fetchNextRequest()).toBeNull(); + + expect(Date.now() - start).toBeLessThan(1000); + }); + + test('picks up requests left in per-domain sub-queues by a previous run', async () => { + const domains = ['example.com']; + + const firstRun = new ThrottlingRequestManager({ inner: await createQueue(), domains }); + await firstRun.addRequest({ url: 'https://example.com/left-behind' }); + expect(await firstRun.getPendingCount()).toBe(1); + + // A restart builds a brand new manager over the same storage backend. + const secondRun = new ThrottlingRequestManager({ inner: await createQueue(), domains }); + + expect(await secondRun.isEmpty()).toBe(false); + expect(await secondRun.isFinished()).toBe(false); + expect(await secondRun.getPendingCount()).toBe(1); + expect((await secondRun.fetchNextRequest())!.url).toBe('https://example.com/left-behind'); + }); + + test('purge empties per-domain sub-queues it has not touched yet', async () => { + const domains = ['example.com']; + + const firstRun = new ThrottlingRequestManager({ inner: await createQueue(), domains }); + await firstRun.addRequest({ url: 'https://example.com/stale' }); + + const subQueue = await RequestQueue.open({ alias: 'throttled-example.com' }); + expect(await subQueue.getPendingCount()).toBe(1); + + const secondRun = new ThrottlingRequestManager({ inner: await createQueue(), domains }); + await secondRun.purge(); + + expect(await subQueue.getPendingCount()).toBe(0); + }); + + describe('stall detection', () => { + const stallingManager = async () => + new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + baseDelaySecs: 0.01, + maxDomainStallSecs: 30, + }); + + /** + * Ages the domain's ongoing run of 429s past the stall threshold. Backdating beats sleeping - a loaded + * CI box cannot race it. + */ + const stallFor = (manager: ThrottlingRequestManager, domain: string) => { + domainState(manager, domain).rateLimitedSince -= 60_000; + }; + + test('gives up on a domain that never lets a request through', async () => { + const manager = await stallingManager(); + await manager.addRequest({ url: 'https://example.com/1' }); + manager.recordDomainDelay('https://example.com/1'); + + await expect(manager.assertNoStalledDomains()).resolves.toBeUndefined(); + + stallFor(manager, 'example.com'); + await expect(manager.assertNoStalledDomains()).rejects.toThrow(PersistentRateLimitError); + }); + + test('a handled request resets the clock', async () => { + const manager = await stallingManager(); + await manager.addRequest({ url: 'https://example.com/1' }); + await manager.addRequest({ url: 'https://example.com/2' }); + manager.recordDomainDelay('https://example.com/1'); + stallFor(manager, 'example.com'); + + await manager.markRequestAsHandled((await pollForNextRequest(manager))!); + + await expect(manager.assertNoStalledDomains()).resolves.toBeUndefined(); + }); + + test('a domain that has run out of work is finished, not stalled', async () => { + const manager = await stallingManager(); + manager.recordDomainDelay('https://example.com/1'); + stallFor(manager, 'example.com'); + + await expect(manager.assertNoStalledDomains()).resolves.toBeUndefined(); + }); + + test('a domain that has been idle for longer than the window is not stalled by its first 429', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + baseDelaySecs: 0.01, + maxDomainStallSecs: 0.05, + }); + await manager.addRequest({ url: 'https://example.com/1' }); + + // The crawl spent longer than the whole stall window elsewhere before this domain was touched. + await sleep(100); + + // The first 429 starts the clock - it does not arrive with the idle time already on it. + manager.recordDomainDelay('https://example.com/1'); + + await expect(manager.assertNoStalledDomains()).resolves.toBeUndefined(); + }); + + test('a domain that was never rate-limited is never stalled', async () => { + const manager = await stallingManager(); + await manager.addRequest({ url: 'https://example.com/1' }); + + await expect(manager.assertNoStalledDomains()).resolves.toBeUndefined(); + }); + + test('a domain that stopped rate-limiting a while ago is being waited out, not stalled', async () => { + const manager = await stallingManager(); + await manager.addRequest({ url: 'https://example.com/1' }); + + // A single old 429, and nothing since - which is what a `Crawl-delay` longer than the stall window + // looks like. The domain is not turning us away, we are keeping our distance from it. + manager.recordDomainDelay('https://example.com/1'); + stallFor(manager, 'example.com'); + domainState(manager, 'example.com').lastRateLimitedAt -= 60_000; + + await expect(manager.assertNoStalledDomains()).resolves.toBeUndefined(); + }); + }); + + test('a crawl-delay does not swallow the 429 backoff', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + }); + + manager.setCrawlDelay('https://example.com/1', 5); + await manager.addRequest({ url: 'https://example.com/1' }); + await manager.addRequest({ url: 'https://example.com/2' }); + + // Dispatching arms the crawl-delay, which used to read as an already-active backoff. + await manager.fetchNextRequest(); + expect(manager.recordDomainDelay('https://example.com/1', 30_000)).toBe(true); + + const state = domainState(manager, 'example.com'); + expect(state.consecutive429Count).toBe(1); + expect(state.backoffUntil).toBeGreaterThan(Date.now() + 25_000); + + // The longer of the two clocks wins, so the domain stays parked. + expect(await manager.fetchNextRequest()).toBeNull(); + }); + + test('concurrent fetches cannot dispatch past the crawl-delay', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + // Holds every dispatch open long enough that the other callers demonstrably run while one is in + // flight, rather than leaving the overlap to microtask ordering. + requestManagerOpener: async (identifier, options) => { + const queue = await RequestQueue.open(identifier, options); + const fetch = queue.fetchNextRequest.bind(queue); + queue.fetchNextRequest = async (...args) => { + await sleep(50); + return fetch(...args); + }; + return queue; + }, + domains: ['example.com'], + }); + + manager.setCrawlDelay('https://example.com/1', 60); + for (let i = 0; i < 5; i++) { + await manager.addRequest({ url: `https://example.com/${i}` }); + } + + // The task loop runs several tasks at once, each pulling its own request. + const fetched = await Promise.all(Array.from({ length: 5 }, async () => manager.fetchNextRequest())); + + expect(fetched.filter(Boolean)).toHaveLength(1); + }); + + test('a domain that hands over nothing does not spend its crawl-delay slot', async () => { + const manager = new ThrottlingRequestManager({ + inner: await createQueue(), + domains: ['example.com'], + }); + + manager.setCrawlDelay('https://example.com/1', 60); + + // Nothing queued yet, so there is no dispatch for the delay to pace. + expect(await manager.fetchNextRequest()).toBeNull(); + + await manager.addRequest({ url: 'https://example.com/1' }); + expect((await manager.fetchNextRequest())!.url).toBe('https://example.com/1'); + }); + + test('setCrawlDelay sets crawl-delay successfully', async () => { + const inner = await createQueue(); + const manager = new ThrottlingRequestManager({ + inner, + domains: ['example.com'], + }); + + manager.setCrawlDelay('https://example.com/1', 0.2); // 0.2 seconds = 200ms + + await manager.addRequest({ url: 'https://example.com/1' }); + await manager.addRequest({ url: 'https://example.com/2' }); + + const req1 = await manager.fetchNextRequest(); + expect(req1!.url).toBe('https://example.com/1'); + + // Dispatching req1 pushes example.com's next slot 200ms out. + const start = Date.now(); + const req2 = await pollForNextRequest(manager); + + expect(Date.now() - start).toBeGreaterThanOrEqual(150); + expect(req2.url).toBe('https://example.com/2'); + }); +});