fix: set the log level per run and answer callers when the run migrates

src/crawlers.ts

@@ -217,8 +217,7 @@ export async function createAndStartSearchCrawler(
         failedRequestHandler: async ({ request }, err) => {
             addTimeMeasureEvent(request.userData!, 'cheerio-failed-request');
             log.error(`Google-search-crawler failed to process request ${request.url}, error ${err.message}`);
-            const errorResponse = { errorMessage: err.message };
-            sendResponseError(request.uniqueKey, JSON.stringify(errorResponse));
+            sendResponseError(request.userData!.responseId, err.message);
         },
     }));
 

src/input.ts

@@ -42,6 +42,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.
@@ -76,6 +78,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),
@@ -93,6 +96,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);
@@ -129,8 +137,6 @@ async function processInputInternal(
         removeCookieWarnings,
     } = input;
 
-    log.setLevel(debugMode ? log.LEVELS.DEBUG : log.LEVELS.INFO);
-
     const contentScraperSettings: ContentScraperSettings = {
         debugMode,
         dynamicContentWaitSecs,

src/main.ts

@@ -4,7 +4,7 @@ import { log } from 'crawlee';
 import { createAndStartContentCrawler, createAndStartSearchCrawler } from './crawlers.js';
 import { processInput, processStandbyInput } from './input.js';
 import { getMiniActor } from './mini-actors.js';
-import { addTimeoutToAllResponses } from './responses.js';
+import { failAllResponsesOnMigration } from './responses.js';
 import { handleSearchNormalMode } from './search.js';
 import { createServer } from './server.js';
 import { findActorTip, storeActorTip } from './tips.js';
@@ -13,12 +13,13 @@ import { isActorStandby } from './utils.js';
 
 await Actor.init();
 
-Actor.on('migrating', () => {
-    addTimeoutToAllResponses(60);
-});
+Actor.on('migrating', failAllResponsesOnMigration);
 
 const originalInput = await Actor.getInput<Partial<Input>>() ?? {} as Input;
 
+// 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()) {
     log.info('Actor is running in the STANDBY mode.');
 

src/request-handler.ts

@@ -198,6 +198,8 @@ async function handleContent(
     const processedHtml = await processHtml(html, baseUrl, settings, $);
     addTimeMeasureEvent(request.userData, `${crawlerType}-process-html`);
 
+    checkTimeoutAndCancelRequest(request, request.userData.responseId);
+
     const isTooLarge = processedHtml.length > settings.maxHtmlCharsToProcess;
     const text = isTooLarge ? load(processedHtml).text() : htmlToText(load(processedHtml).html());
 

src/responses.ts

@@ -163,19 +163,13 @@ export function sendResponseIfFinished(responseId: string) {
         sendResponseOk(responseId, sortResultsByRank(res));
     }
 }
+
 /**
- * Add timeout to all responses when actor is migrating (source: SuperScraper).
+ * 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 const addTimeoutToAllResponses = (timeoutSeconds = 60) => {
-    const migrationErrorMessage = {
-        errorMessage: `Actor had to migrate to another server. Please, retry your request.`,
-    };
-
-    const responseKeys = Object.keys(responseData);
-
-    for (const key of responseKeys) {
-        setTimeout(() => {
-            sendResponseError(key, JSON.stringify(migrationErrorMessage));
-        }, timeoutSeconds * 1000);
+export function failAllResponsesOnMigration() {
+    for (const responseId of [...responseData.keys()]) {
+        sendResponseError(responseId, 'Actor had to migrate to another server. Please, retry your request.');
     }
-};
+}

tests/input.test.ts

@@ -1,7 +1,17 @@
+import { Actor } from 'apify';
 import { log } from 'crawlee';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 
-import { warnIfLowMemoryForPlaywright } from '../src/input.js';
+import { processInput, warnIfLowMemoryForPlaywright } from '../src/input.js';
+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(() => {
@@ -32,3 +42,29 @@ describe('warnIfLowMemoryForPlaywright', () => {
         expect(warning).not.toHaveBeenCalled();
     });
 });
+
+describe('the log level', () => {
+    it('ignores a caller\'s debugMode, which would switch logging for everyone sharing the run', async () => {
+        log.setLevel(log.LEVELS.INFO);
+
+        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

@@ -0,0 +1,54 @@
+import { describe, expect, it } from 'vitest';
+
+import { ContentCrawlerStatus } from '../src/const.js';
+import { createAndStartSearchCrawler } from '../src/crawlers.js';
+import {
+    addEmptyResultToResponse,
+    addResultToResponse,
+    createResponsePromise,
+    failAllResponsesOnMigration,
+} from '../src/responses.js';
+
+describe('a run that migrates to another server', () => {
+    it('tells a caller still waiting to retry, rather than leaving it on its own timeout', async () => {
+        const waiting = createResponsePromise('nothing-finished', 300);
+
+        failAllResponsesOnMigration();
+
+        await expect(waiting).rejects.toThrow('Actor had to migrate to another server. Please, retry your request.');
+    });
+
+    // 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';
+        addEmptyResultToResponse('one-page-finished', { url, uniqueKey: url } as never);
+        addResultToResponse('one-page-finished', url, {
+            crawl: { requestStatus: ContentCrawlerStatus.HANDLED },
+            metadata: { url },
+            searchResult: { rank: 1 },
+        } as never);
+
+        failAllResponsesOnMigration();
+
+        await expect(waiting).resolves.toMatchObject([{ metadata: { url } }]);
+    });
+});
+
+describe('a search that fails', () => {
+    it('tells the caller why, instead of leaving them to wait out their own timeout', async () => {
+        const waiting = createResponsePromise('search-failed', 300);
+        const { crawler } = await createAndStartSearchCrawler({}, false);
+        const { failedRequestHandler } = crawler as unknown as {
+            failedRequestHandler: (context: unknown, err: Error) => Promise<void>;
+        };
+
+        await failedRequestHandler(
+            // The request's own key is not the response's: the response is found by `responseId`.
+            { request: { url: 'https://www.google.com/search?q=x', uniqueKey: 'some-other-key', userData: { responseId: 'search-failed' } } },
+            new Error('Google refused the request'),
+        );
+
+        await expect(waiting).rejects.toThrow('Google refused the request');
+    });
+});