fix: size the shared crawler pool from the run, not from a request

Author: nikitachapovskii-dev

src/input.ts

@@ -46,6 +46,8 @@ async function getProxyConfiguration(proxyOptions: ProxyOptions) {
 
 const DEFAULT_PROXY_OPTIONS = ragWebBrowserInputSchema.properties.proxyConfiguration.default as ProxyOptions;
 
+let runDesiredConcurrency = ragWebBrowserInputSchema.properties.desiredConcurrency.default;
+
 /**
  * Resolves the proxy from the request Crawlee is about to send, so one crawler serves callers that
  * each asked for a different proxy. This is what keeps the Standby crawler set at three.
@@ -80,6 +82,7 @@ async function validateProxyConfiguration(proxyOptions: ProxyOptions, abortOnFai
 export async function processStandbyInput(originalInput: Partial<Input>) {
     const { input, searchCrawlerOptions, contentScraperSettings } = await processInputInternal(originalInput, true);
 
+    runDesiredConcurrency = input.desiredConcurrency;
     await validateProxyConfiguration(input.proxyConfiguration, true);
     const contentCrawlerOptions: ContentCrawlerOptions[] = [
         createPlaywrightCrawlerOptions(input),
@@ -97,6 +100,11 @@ export async function processInput(originalInput: Partial<Input>) {
 
     // In Standby a proxy problem must fail one request; in Normal mode there is nothing to serve.
     await validateProxyConfiguration(input.proxyConfiguration, !isActorStandby());
+
+    // A Standby request builds these options whenever no crawler is cached yet, so whichever caller
+    // got there first would size the pool every later caller shares. The run sizes it instead.
+    if (isActorStandby()) input.desiredConcurrency = runDesiredConcurrency;
+
     const contentCrawlerOptions: ContentCrawlerOptions = input.scrapingTool === 'raw-http'
         ? createCheerioCrawlerOptions(input, false)
         : createPlaywrightCrawlerOptions(input, false);

src/main.ts

@@ -17,8 +17,7 @@ Actor.on('migrating', failAllResponsesOnMigration);
 
 const originalInput = await Actor.getInput<Partial<Input>>() ?? {} as Input;
 
-// The log level is process-wide, so it follows the Actor's own input. In Standby `debugMode` arrives
-// per request, where honoring it would switch logging for every other caller sharing the run.
+// Process-wide, so it follows the run's input: a caller's `debugMode` must not reach other callers.
 log.setLevel(originalInput.debugMode ? log.LEVELS.DEBUG : log.LEVELS.INFO);
 
 if (isActorStandby()) {

src/responses.ts

@@ -165,9 +165,8 @@ export function sendResponseIfFinished(responseId: string) {
 }
 
 /**
- * Tells everyone still waiting that this run is moving to another server, so they retry instead of
- * waiting out their own timeout. Sent at once rather than on a timer, because the announcement is
- * the last moment we know the process is still here to answer with.
+ * Tells everyone still waiting to retry elsewhere. Sent at once rather than on a timer, because the
+ * announcement is the last moment we know the process is still here to answer with.
  */
 export function failAllResponsesOnMigration() {
     for (const responseId of [...responseData.keys()]) {

tests/input.test.ts

@@ -1,3 +1,4 @@
+import { Actor } from 'apify';
 import { log } from 'crawlee';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 
@@ -6,6 +7,12 @@ import { parseParameters } from '../src/utils.js';
 
 process.env.ACTOR_FULL_NAME = 'apify/rag-web-browser';
 
+// A custom proxy keeps `Actor.createProxyConfiguration` off the network.
+const CALLER_PROXY = encodeURIComponent(JSON.stringify({
+    useApifyProxy: false,
+    proxyUrls: ['http://caller.invalid:8000'],
+}));
+
 describe('warnIfLowMemoryForPlaywright', () => {
     afterEach(() => {
         vi.restoreAllMocks();
@@ -38,15 +45,26 @@ describe('warnIfLowMemoryForPlaywright', () => {
 
 describe('the log level', () => {
     it('ignores a caller\'s debugMode, which would switch logging for everyone sharing the run', async () => {
-        // A custom proxy keeps `Actor.createProxyConfiguration` off the network.
-        const proxy = encodeURIComponent(JSON.stringify({
-            useApifyProxy: false,
-            proxyUrls: ['http://log-level.invalid:8000'],
-        }));
         log.setLevel(log.LEVELS.INFO);
 
-        await processInput(parseParameters(`?query=hello&debugMode=true&proxyConfiguration=${proxy}`));
+        await processInput(parseParameters(`?query=hello&debugMode=true&proxyConfiguration=${CALLER_PROXY}`));
 
         expect(log.getLevel()).toBe(log.LEVELS.INFO);
     });
 });
+
+describe('the shared crawler pool', () => {
+    afterEach(() => {
+        vi.restoreAllMocks();
+    });
+
+    it('is sized by the run, not by whichever caller happens to build it', async () => {
+        vi.spyOn(Actor, 'getEnv').mockReturnValue({ metaOrigin: 'STANDBY' } as never);
+
+        const { contentCrawlerOptions } = await processInput(
+            parseParameters(`?query=hello&desiredConcurrency=17&proxyConfiguration=${CALLER_PROXY}`),
+        );
+
+        expect(contentCrawlerOptions.crawlerOptions.autoscaledPoolOptions?.desiredConcurrency).toBe(5);
+    });
+});

tests/responses.test.ts

@@ -17,8 +17,7 @@ describe('a run that migrates to another server', () => {
         await expect(waiting).rejects.toThrow('Actor had to migrate to another server. Please, retry your request.');
     });
 
-    // This is what makes it safe to answer the moment the migration is announced, instead of giving
-    // the crawl a grace period the process may not live long enough to honour.
+    // This is what makes it safe to answer at once instead of giving the crawl a grace period.
     it('still returns the results a caller already has', async () => {
         const waiting = createResponsePromise('one-page-finished', 300);
         const url = 'https://example.com/finished';