Prevent maxNewRequests from blocking on async producers

Exact `requestsOverLimit` reporting for async iterables currently requires
fully draining the producer after the budget is already exhausted. That
turns a bounded enqueue operation into a blocking one when the producer
stalls. This change keeps exact leftover reporting for materialized inputs
(and for callers that explicitly opt into waiting), while letting the
default async-iterable path return as soon as the budget is reached.

Constraint: Preserve exact leftover reporting for arrays and for callers that explicitly set waitForAllRequestsToBeAdded
Rejected: Remove requestsOverLimit entirely for every input type | too broad and breaks useful existing behavior for materialized sources
Confidence: medium
Scope-risk: moderate
Reversibility: clean
Directive: If the requestsOverLimit contract changes again, keep async-producer liveness and exact leftover reporting as separate concerns
Tested: `node node_modules/vitest/vitest.mjs run test/core/crawlers/basic_crawler.test.ts -t 'addRequestsBatched with maxNewRequests should not wait for an async iterable beyond the remaining budget'`; `node node_modules/vitest/vitest.mjs run test/core/crawlers/basic_crawler.test.ts -t 'addRequestsBatched with maxNewRequests should correctly report requestsOverLimit for generator input'`; `node node_modules/vitest/vitest.mjs run test/core/crawlers/basic_crawler.test.ts -t 'addRequestsBatched with maxNewRequests should correctly report requestsOverLimit for array input'`
Not-tested: Full test suite; behavior for custom iterable implementations beyond the covered array/async-generator cases

Author: shaun0927

packages/core/src/storages/request_provider.ts

@@ -456,6 +456,8 @@ export abstract class RequestProvider implements IStorage, IRequestManager {
         }
 
         const { batchSize = 1000, waitBetweenBatchesMillis = 1000, maxNewRequests } = options;
+        const shouldCollectExactRequestsOverLimit =
+            maxNewRequests !== undefined && (!isAsyncIterable(requests) || options.waitForAllRequestsToBeAdded);
 
         let remainingBudget = maxNewRequests ?? Infinity;
         const requestsOverLimit: Source[] = [];
@@ -507,8 +509,9 @@ export abstract class RequestProvider implements IStorage, IRequestManager {
         };
 
         /**
-         * Build the final result. When maxNewRequests is set, drains any remaining items
-         * from the underlying request iterator into requestsOverLimit.
+         * Build the final result. When maxNewRequests is set for a materialized source (or when the caller
+         * explicitly asked to wait for all requests), 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 —
@@ -521,7 +524,7 @@ export abstract class RequestProvider implements IStorage, IRequestManager {
             waitForAllRequestsToBeAdded: Promise<ProcessedRequest[]>,
             unconsumedIterator: AsyncGenerator<RequestOptions>,
         ): Promise<AddRequestsBatchedResult> => {
-            if (maxNewRequests !== undefined) {
+            if (shouldCollectExactRequestsOverLimit) {
                 for await (const request of unconsumedIterator) {
                     requestsOverLimit.push(request);
                 }
@@ -561,8 +564,8 @@ export abstract class RequestProvider implements IStorage, IRequestManager {
             this.inProgressRequestBatchCount -= 1;
         });
 
-        // When maxNewRequests is set, we must wait for all batches so we can accurately report skipped requests.
-        if (options.waitForAllRequestsToBeAdded || maxNewRequests !== undefined) {
+        // When exact requestsOverLimit reporting is requested, we need to await the background work.
+        if (options.waitForAllRequestsToBeAdded || shouldCollectExactRequestsOverLimit) {
             addedRequests.push(...(await promise));
         }
 
@@ -1031,8 +1034,10 @@ export interface AddRequestsBatchedOptions extends RequestQueueOperationOptions
      *
      * This is useful in combination with `maxRequestsPerCrawl` to avoid duplicate URLs consuming the budget.
      *
-     * **Note:** Setting this option implicitly enables {@apilink AddRequestsBatchedOptions.waitForAllRequestsToBeAdded|`waitForAllRequestsToBeAdded`},
-     * since all batches must complete before leftover requests can be accurately reported.
+     * **Note:** For async iterables, exact `requestsOverLimit` reporting is only available when
+     * {@apilink AddRequestsBatchedOptions.waitForAllRequestsToBeAdded|`waitForAllRequestsToBeAdded`}
+     * is set to `true`. Otherwise, Crawlee returns as soon as the budget is exhausted without draining
+     * the producer beyond that point.
      */
     maxNewRequests?: number;
 }
@@ -1061,6 +1066,10 @@ export interface AddRequestsBatchedResult {
      * Requests from the input that were not added to the queue because the
      * {@apilink AddRequestsBatchedOptions.maxNewRequests|`maxNewRequests`} budget was reached.
      * Empty when `maxNewRequests` is not set.
+     *
+     * For async iterables, exact leftovers are only reported when
+     * {@apilink AddRequestsBatchedOptions.waitForAllRequestsToBeAdded|`waitForAllRequestsToBeAdded`}
+     * is enabled.
      */
     requestsOverLimit?: Source[];
 }

test/core/crawlers/basic_crawler.test.ts

@@ -1947,7 +1947,10 @@ describe('BasicCrawler', () => {
                 yield { url: 'http://example.com/e' };
             }
 
-            const result = await queue.addRequestsBatched(urls(), { maxNewRequests: 2 });
+            const result = await queue.addRequestsBatched(urls(), {
+                maxNewRequests: 2,
+                waitForAllRequestsToBeAdded: true,
+            });
 
             const addedUrls = result.addedRequests.filter((r) => !r.wasAlreadyPresent).map((r) => r.uniqueKey);
 
@@ -1957,6 +1960,40 @@ describe('BasicCrawler', () => {
             expect(overLimitUrls).toHaveLength(3);
         });
 
+        test('addRequestsBatched with maxNewRequests should not wait for an async iterable beyond the remaining budget', async () => {
+            const queue = await RequestQueue.open();
+            let consumed = 0;
+            let releaseBlockedRequest = () => {};
+            const blockedRequest = new Promise<void>((resolve) => {
+                releaseBlockedRequest = resolve;
+            });
+
+            async function* urls() {
+                consumed += 1;
+                yield { url: 'http://example.com/a' };
+
+                consumed += 1;
+                await blockedRequest;
+                yield { url: 'http://example.com/b' };
+            }
+
+            const pendingResult = queue.addRequestsBatched(urls(), {
+                maxNewRequests: 1,
+                waitBetweenBatchesMillis: 0,
+            });
+
+            const raced = await Promise.race([
+                pendingResult.then(() => 'resolved'),
+                new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)),
+            ]);
+
+            expect(raced).toBe('resolved');
+            expect(consumed).toBe(1);
+
+            releaseBlockedRequest();
+            await pendingResult;
+        });
+
         test('should not count duplicate URLs toward maxRequestsPerCrawl limit (enqueueLinks)', async () => {
             const requestQueue = await RequestQueue.open();