fix(core): cap retries for unprocessed requests in addRequestsBatched

When the request queue permanently rejects a request (e.g. a 400 from the
platform batch-add API for a malformed `userData` shape), the recursive
helper kept resubmitting the unprocessed requests forever, hanging the
crawler with no actionable error.

Bound the recursion to a small number of consecutive no-progress attempts;
once exhausted, the remaining unprocessed requests are skipped with a
warning. Any progress resets the counter so transient backpressure is still
retried.

Closes #3764

Author: B4nan

packages/core/src/storages/request_provider.ts

@@ -38,6 +38,12 @@ import { getRequestId, purgeDefaultStorages, QUERY_HEAD_MIN_LENGTH } from './uti
 
 export type RequestsLike = AsyncIterable<Source | string> | Iterable<Source | string> | (Source | string)[];
 
+/**
+ * Maximum number of consecutive batch-add attempts that make no progress before the remaining
+ * unprocessed requests are skipped, so permanently rejected requests don't retry forever.
+ */
+const MAX_UNPROCESSED_REQUESTS_RETRIES = 3;
+
 /**
  * Represents a provider of requests/URLs to crawl.
  */
@@ -472,12 +478,29 @@ export abstract class RequestProvider implements IStorage, IRequestManager {
         );
         const chunksIterator = chunks[Symbol.asyncIterator]();
 
-        const attemptToAddToQueueAndAddAnyUnprocessed = async (providedRequests: Source[], cache = true) => {
+        const attemptToAddToQueueAndAddAnyUnprocessed = async (
+            providedRequests: Source[],
+            cache = true,
+            unsuccessfulAttempts = 0,
+        ) => {
             const resultsToReturn: ProcessedRequest[] = [];
             const apiResult = await this.addRequests(providedRequests, { forefront: options.forefront, cache });
             resultsToReturn.push(...apiResult.processedRequests);
 
             if (apiResult.unprocessedRequests.length) {
+                // Count attempts that make no progress, so permanently rejected requests (e.g. a malformed
+                // `userData` shape causing a 400) don't loop forever. Any progress resets the counter.
+                const attempts = apiResult.processedRequests.length ? 0 : unsuccessfulAttempts + 1;
+
+                if (attempts >= MAX_UNPROCESSED_REQUESTS_RETRIES) {
+                    this.log.warning(
+                        `Some requests were consistently rejected by the request queue and will be skipped after ${MAX_UNPROCESSED_REQUESTS_RETRIES} attempts. This usually means the request data is malformed (e.g. an invalid 'userData' shape).`,
+                        { unprocessedRequests: apiResult.unprocessedRequests },
+                    );
+
+                    return resultsToReturn;
+                }
+
                 await sleep(waitBetweenBatchesMillis);
 
                 resultsToReturn.push(
@@ -486,6 +509,7 @@ export abstract class RequestProvider implements IStorage, IRequestManager {
                             (r) => !apiResult.processedRequests.some((pr) => pr.uniqueKey === r.uniqueKey),
                         ),
                         false,
+                        attempts,
                     )),
                 );
             }

test/core/storages/request_queue.test.ts

@@ -278,6 +278,30 @@ describe('RequestQueue remote', () => {
         expect(mockAddRequests).toBeCalledWith([requestB, requestC], { forefront: true });
     });
 
+    test('addRequestsBatched does not retry permanently unprocessed requests forever', async () => {
+        const queue = new RequestQueue({ id: 'unprocessed-requests', client: storageClient });
+        const mockAddRequests = vitest.spyOn(queue.client, 'batchAddRequests');
+
+        const requestOptions = { url: 'http://example.com/bad' };
+        const request = new Request(requestOptions);
+
+        // Simulate the platform permanently rejecting the request (e.g. a 400 due to a malformed `userData` shape):
+        // it is always reported back as unprocessed.
+        mockAddRequests.mockResolvedValue({
+            processedRequests: [],
+            unprocessedRequests: [{ uniqueKey: request.uniqueKey, url: request.url, method: 'GET' }],
+        });
+
+        const logWarningSpy = vitest.spyOn(queue.log, 'warning');
+
+        const result = await queue.addRequestsBatched([requestOptions], { waitBetweenBatchesMillis: 0 });
+
+        // Must not hang: it gives up after a bounded number of attempts and warns about the skipped requests.
+        expect(result.addedRequests).toHaveLength(0);
+        expect(logWarningSpy).toHaveBeenCalled();
+        expect(mockAddRequests.mock.calls.length).toBeLessThan(20);
+    });
+
     test('should cache new requests locally', async () => {
         const queue = new RequestQueue({ id: 'some-id', client: storageClient });