fix: let one crawler set serve every proxy a caller asks for

Author: nikitachapovskii-dev

src/crawlers.ts

@@ -1,4 +1,3 @@
-import { createHash } from 'node:crypto';
 import { readFile } from 'node:fs/promises';
 
 import { ImpitHttpClient } from '@crawlee/impit-client';
@@ -33,15 +32,15 @@ import type {
     ContentCrawlerOptions,
     ContentCrawlerUserData,
     Output,
-    ProxyOptions,
-    SearchCrawlerOptions,
     SearchCrawlerUserData,
 } from './types.js';
 import { addTimeMeasureEvent, createRequest, createSearchRequest } from './utils.js';
 
 type Crawler = CheerioCrawler | PlaywrightCrawler;
 
-const crawlers = new Map<string, Crawler>();
+// A kind is all that distinguishes one crawler from another: nothing a caller can ask for reaches a
+// crawler constructor any more, so the set cannot grow past one crawler per kind.
+const crawlers = new Map<CrawlerKind, Promise<Crawler>>();
 const client = new MemoryStorage({ persistStorage: false });
 
 const contentCrawlerHttpClient = new ImpitHttpClient({
@@ -67,41 +66,43 @@ async function getGhosteryBlocker(): Promise<PlaywrightBlocker | undefined> {
     }
 }
 
-/**
- * Identifies a crawler in the `crawlers` cache, and names its request queue. Keyed off the caller's
- * proxy options rather than the built `ProxyConfiguration`, whose child logger snapshots the log level
- * and whose URL rotation state mutates - both of which kept changing the key. Hashed because the key
- * is logged and used as a queue name, while the options can hold credentials.
- */
-export function getCrawlerKey(kind: CrawlerKind, proxyOptions: ProxyOptions): string {
-    const hash = createHash('sha1').update(JSON.stringify(proxyOptions)).digest('hex');
-    return `${kind}-${hash.slice(0, 12)}`;
-}
-
 /** How many crawlers are alive. Exported so tests can assert that requests do not fork new ones. */
 export function getCrawlerCount() {
     return crawlers.size;
 }
 
-async function getOrCreateCrawler(key: string, kind: CrawlerKind, startCrawler: boolean, build: () => Promise<Crawler>) {
-    const cached = crawlers.get(key);
+/**
+ * Holds the pending crawler rather than the built one: the server is listening before the crawlers
+ * started at boot have finished building, so two concurrent requests could otherwise each build one
+ * and orphan all but the last. A crawler that fails to build or to run is dropped, so the cache can
+ * never hand out a dead one.
+ */
+async function getOrCreateCrawler(kind: CrawlerKind, startCrawler: boolean, build: () => Promise<Crawler>) {
+    const cached = crawlers.get(kind);
     if (cached) {
         return cached;
     }
 
-    log.info(`Creating new ${kind} crawler with key ${key}`);
-    const crawler = await build();
-    if (startCrawler) {
-        crawler.run().then(
-            () => log.warning(`Crawler ${kind} has finished`),
-            // eslint-disable-next-line @typescript-eslint/no-empty-function
-            () => { },
-        );
-        log.info(`Crawler ${kind} has started 💪🏼`);
-    }
-    crawlers.set(key, crawler);
+    const pending = (async () => {
+        log.info(`Creating new ${kind} crawler`);
+        const crawler = await build();
+        if (startCrawler) {
+            crawler.run().then(
+                () => log.warning(`Crawler ${kind} has finished`),
+                (err) => {
+                    log.error(`Crawler ${kind} failed to run: ${err instanceof Error ? err.message : String(err)}`);
+                    crawlers.delete(kind);
+                },
+            );
+            log.info(`Crawler ${kind} has started 💪🏼`);
+        }
+        return crawler;
+    })();
+
+    crawlers.set(kind, pending);
+    pending.catch(() => crawlers.delete(kind));
     log.info(`Number of crawlers ${crawlers.size}`);
-    return crawler;
+    return pending;
 }
 
 /**
@@ -110,14 +111,15 @@ async function getOrCreateCrawler(key: string, kind: CrawlerKind, startCrawler:
 export const addContentCrawlRequest = async (
     request: RequestOptions<ContentCrawlerUserData>,
     responseId: string,
-    contentCrawlerKey: string,
+    contentCrawlerKey: CrawlerKind,
 ) => {
-    const crawler = crawlers.get(contentCrawlerKey);
-    if (!crawler) {
+    const pending = crawlers.get(contentCrawlerKey);
+    if (!pending) {
         log.error(`Content crawler not found: ${contentCrawlerKey}`);
         return;
     }
 
+    const crawler = await pending;
     const name = crawler instanceof PlaywrightCrawler ? 'playwright' : 'cheerio';
     try {
         await crawler.requestQueue!.addRequest(request);
@@ -135,13 +137,12 @@ export const addContentCrawlRequest = async (
  * A crawler won't be created if it already exists.
  */
 export async function createAndStartSearchCrawler(
-    searchCrawlerOptions: SearchCrawlerOptions,
+    searchCrawlerOptions: CheerioCrawlerOptions,
     startCrawler = true,
 ) {
-    const { crawlerOptions, proxyOptions } = searchCrawlerOptions;
-    const key = getCrawlerKey('search', proxyOptions);
-    const crawler = await getOrCreateCrawler(key, 'search', startCrawler, async () => new CheerioCrawler({
-        ...crawlerOptions,
+    const key: CrawlerKind = 'search';
+    const crawler = await getOrCreateCrawler(key, startCrawler, async () => new CheerioCrawler({
+        ...searchCrawlerOptions,
         requestQueue: await RequestQueue.open(key, { storageClient: client }),
         requestHandler: async ({ request, $: _$, addRequests }: CheerioCrawlingContext<SearchCrawlerUserData>) => {
             // NOTE: we need to cast this to fix `cheerio` type errors
@@ -189,7 +190,6 @@ export async function createAndStartSearchCrawler(
                         collectedResults: deduplicated,
                         currentPage: nextPage,
                     },
-                    proxyOptions,
                     nextOffset,
                 );
                 await addRequests([nextRequest]);
@@ -208,6 +208,7 @@ export async function createAndStartSearchCrawler(
                         result,
                         responseId,
                         request.userData.contentScraperSettings!,
+                        request.userData.contentProxyOptions,
                         request.userData.timeMeasures!,
                         actorRequestId,
                     );
@@ -235,16 +236,15 @@ export async function createAndStartContentCrawler(
     contentCrawlerOptions: ContentCrawlerOptions,
     startCrawler = true,
 ) {
-    const { type: crawlerType, crawlerOptions, proxyOptions } = contentCrawlerOptions;
+    const { type: crawlerType, crawlerOptions } = contentCrawlerOptions;
 
-    const key = getCrawlerKey(crawlerType, proxyOptions);
-    const crawler = await getOrCreateCrawler(key, crawlerType, startCrawler, async () => (
+    const crawler = await getOrCreateCrawler(crawlerType, startCrawler, async () => (
         crawlerType === ContentCrawlerTypes.PLAYWRIGHT
-            ? createPlaywrightContentCrawler(crawlerOptions, key)
-            : createCheerioContentCrawler(crawlerOptions, key)
+            ? createPlaywrightContentCrawler(crawlerOptions, crawlerType)
+            : createCheerioContentCrawler(crawlerOptions, crawlerType)
     ));
 
-    return { key, crawler };
+    return { key: crawlerType, crawler };
 }
 
 /**
@@ -326,14 +326,15 @@ async function createCheerioContentCrawler(
  */
 export const addSearchRequest = async (
     request: RequestOptions<ContentCrawlerUserData>,
-    searchCrawlerKey: string,
+    searchCrawlerKey: CrawlerKind,
 ) => {
-    const crawler = crawlers.get(searchCrawlerKey);
-    if (!crawler) {
+    const pending = crawlers.get(searchCrawlerKey);
+    if (!pending) {
         log.error(`Search crawler not found: ${searchCrawlerKey}`);
         return;
     }
 
+    const crawler = await pending;
     addTimeMeasureEvent(request.userData!, 'before-cheerio-queue-add');
     await crawler.requestQueue!.addRequest(request);
     log.info(`Added request to cheerio-google-search-crawler: ${request.url}`);

src/input.ts

@@ -1,7 +1,7 @@
 import type { ProxyConfigurationOptions } from 'apify';
 import { Actor } from 'apify';
-import type { ProxyConfiguration } from 'crawlee';
-import { BrowserName, log } from 'crawlee';
+import type { CheerioCrawlerOptions } from 'crawlee';
+import { BrowserName, log, ProxyConfiguration } from 'crawlee';
 import { firefox } from 'playwright';
 
 import ragWebBrowserInputSchema from '../actors/apify_rag-web-browser/.actor/input_schema.json' with { type: 'json' };
@@ -11,17 +11,17 @@ import { blockMediaRequests } from './media.js';
 import { getMiniActor } from './mini-actors.js';
 import type {
     ContentCrawlerOptions,
+    ContentCrawlerUserData,
     ContentScraperSettings,
     Input,
     OutputFormats,
     ProxyOptions,
     RagWebBrowserInput,
     ScrapingTool,
-    SearchCrawlerOptions,
     SERPProxyGroup,
     UrlToMarkdownInput,
 } from './types.js';
-import { abortRun } from './utils.js';
+import { abortRun, isActorStandby } from './utils.js';
 
 const proxyConfigurations = new Map<string, Promise<ProxyConfiguration | undefined>>();
 
@@ -44,11 +44,43 @@ async function getProxyConfiguration(proxyOptions: ProxyOptions) {
     return configuration;
 }
 
-async function createContentProxyConfiguration(proxyOptions: ProxyOptions) {
+const DEFAULT_PROXY_OPTIONS = ragWebBrowserInputSchema.properties.proxyConfiguration.default as ProxyOptions;
+
+/**
+ * The configuration every crawler is built with. It resolves the proxy from the request Crawlee is
+ * about to send rather than from the input the crawler was built with, so one crawler serves callers
+ * that each asked for a different proxy - which is what keeps the Standby crawler set at three
+ * however many callers there are.
+ *
+ * Crawlee calls this per request for the HTTP crawlers and per page for the browser one. The browser
+ * one only honours it because its launch context sets `useIncognitoPages`; without that, BrowserPool
+ * hands every request whichever browser is free and they all go through the first browser's proxy.
+ */
+const requestProxyConfiguration = new ProxyConfiguration({
+    newUrlFunction: async (sessionId, options) => {
+        // Crawlee also resolves a proxy when it launches a browser, with no request in hand.
+        const userData = options?.request?.userData as Partial<ContentCrawlerUserData> | undefined;
+        const configuration = await getProxyConfiguration(userData?.proxyOptions ?? DEFAULT_PROXY_OPTIONS);
+
+        // Crawlee reads this off the configuration, right after this call, to decide whether to
+        // ignore TLS errors. Keep it in step with whichever configuration served the request.
+        requestProxyConfiguration.isManInTheMiddle = configuration?.isManInTheMiddle ?? false;
+
+        return (await configuration?.newUrl(sessionId)) ?? null;
+    },
+});
+
+/**
+ * Builds the caller's proxy configuration so that an unusable one is reported now, rather than as a
+ * run of failed pages later. The crawlers themselves route through `requestProxyConfiguration`.
+ */
+async function validateProxyConfiguration(proxyOptions: ProxyOptions, abortOnFailure: boolean) {
     try {
-        return await getProxyConfiguration(proxyOptions);
+        await getProxyConfiguration(proxyOptions);
     } catch (e) {
-        return abortRun(`Cannot use Apify Proxy for scraping the target pages: ${(e as Error).message}`);
+        const message = `Cannot use Apify Proxy for scraping the target pages: ${(e as Error).message}`;
+        if (!abortOnFailure) throw new Error(message);
+        await abortRun(message);
     }
 }
 
@@ -59,10 +91,10 @@ async function createContentProxyConfiguration(proxyOptions: ProxyOptions) {
 export async function processStandbyInput(originalInput: Partial<Input>) {
     const { input, searchCrawlerOptions, contentScraperSettings } = await processInputInternal(originalInput, true);
 
-    const proxy = await createContentProxyConfiguration(input.proxyConfiguration);
+    await validateProxyConfiguration(input.proxyConfiguration, true);
     const contentCrawlerOptions: ContentCrawlerOptions[] = [
-        createPlaywrightCrawlerOptions(input, proxy),
-        createCheerioCrawlerOptions(input, proxy),
+        createPlaywrightCrawlerOptions(input),
+        createCheerioCrawlerOptions(input),
     ];
 
     return { input, searchCrawlerOptions, contentCrawlerOptions, contentScraperSettings };
@@ -74,10 +106,12 @@ export async function processStandbyInput(originalInput: Partial<Input>) {
 export async function processInput(originalInput: Partial<Input>) {
     const { input, searchCrawlerOptions, contentScraperSettings } = await processInputInternal(originalInput);
 
-    const proxy = await createContentProxyConfiguration(input.proxyConfiguration);
+    // In Standby this runs once per request, where a proxy problem must fail that request alone. In
+    // Normal mode it is the run's startup, where there is nothing to serve and the run ends.
+    await validateProxyConfiguration(input.proxyConfiguration, !isActorStandby());
     const contentCrawlerOptions: ContentCrawlerOptions = input.scrapingTool === 'raw-http'
-        ? createCheerioCrawlerOptions(input, proxy, false)
-        : createPlaywrightCrawlerOptions(input, proxy, false);
+        ? createCheerioCrawlerOptions(input, false)
+        : createPlaywrightCrawlerOptions(input, false);
 
     return { input, searchCrawlerOptions, contentCrawlerOptions, contentScraperSettings };
 }
@@ -91,7 +125,7 @@ async function processInputInternal(
 ) {
     const miniActor = getMiniActor();
     let input: Input;
-    let searchCrawlerOptions: SearchCrawlerOptions = { crawlerOptions: {}, proxyOptions: {} };
+    let searchCrawlerOptions: CheerioCrawlerOptions = {};
 
     if (miniActor.runsSearch) {
         const processedRagWebBrowserInput = await processRagWebBrowserInput(
@@ -130,7 +164,7 @@ async function processInputInternal(
 async function processRagWebBrowserInput(input: Partial<RagWebBrowserInput>, standbyInit: boolean):
     Promise<{
         validatedRagBrowserInput: RagWebBrowserInput;
-        searchCrawlerOptions: SearchCrawlerOptions
+        searchCrawlerOptions: CheerioCrawlerOptions
     }> {
     /* eslint-disable no-param-reassign */
 
@@ -199,15 +233,11 @@ async function processRagWebBrowserInput(input: Partial<RagWebBrowserInput>, sta
         input.dynamicContentWaitSecs = Math.round(input.requestTimeoutSecs / 2);
     }
 
-    const proxyOptions: ProxyOptions = { groups: [input.serpProxyGroup], checkAccess: false };
-    const searchCrawlerOptions: SearchCrawlerOptions = {
-        crawlerOptions: {
-            keepAlive: standbyInit,
-            maxRequestRetries: CRAWLER_MAX_REQUEST_RETRIES,
-            proxyConfiguration: await getProxyConfiguration(proxyOptions),
-            autoscaledPoolOptions: { desiredConcurrency: 1 },
-        },
-        proxyOptions,
+    const searchCrawlerOptions: CheerioCrawlerOptions = {
+        keepAlive: standbyInit,
+        maxRequestRetries: CRAWLER_MAX_REQUEST_RETRIES,
+        proxyConfiguration: requestProxyConfiguration,
+        autoscaledPoolOptions: { desiredConcurrency: 1 },
     };
     const validatedRagBrowserInput = validateAndFillInput(input) as RagWebBrowserInput;
     return {
@@ -248,20 +278,27 @@ async function processUrlToMarkdownInput(input: Partial<UrlToMarkdownInput>): Pr
 
 function createPlaywrightCrawlerOptions(
     input: Input,
-    proxy: ProxyConfiguration | undefined,
     keepAlive = true,
 ): ContentCrawlerOptions {
     return {
         type: ContentCrawlerTypes.PLAYWRIGHT,
-        proxyOptions: input.proxyConfiguration,
         crawlerOptions: {
             headless: true,
             keepAlive,
+            // The session pool is shared by every caller, so a session's cookies and its Apify Proxy
+            // session id would be reused across requests that went out through different proxies -
+            // one caller's cookies riding another's IP. A request here is one page, which gains
+            // nothing from a session, so there is none.
+            useSessionPool: false,
             maxRequestRetries: CRAWLER_MAX_REQUEST_RETRIES,
-            proxyConfiguration: proxy,
+            proxyConfiguration: requestProxyConfiguration,
             requestHandlerTimeoutSecs: CRAWLER_REQUEST_HANDLER_TIMEOUT_SECS,
             launchContext: {
                 launcher: firefox,
+                // Gives each page its own browser context, which is what lets a page carry the proxy
+                // its request asked for. Without it BrowserPool reuses whichever browser is free and
+                // every request goes through the proxy the first one launched with - silently.
+                useIncognitoPages: true,
             },
             preNavigationHooks: [
                 async ({ page }) => {
@@ -289,17 +326,16 @@ function createPlaywrightCrawlerOptions(
 
 function createCheerioCrawlerOptions(
     input: Input,
-    proxy: ProxyConfiguration | undefined,
     keepAlive = true,
 ): ContentCrawlerOptions {
     return {
         type: ContentCrawlerTypes.CHEERIO,
-        proxyOptions: input.proxyConfiguration,
         crawlerOptions: {
             keepAlive,
             maxRequestRetries: CRAWLER_MAX_REQUEST_RETRIES,
-            proxyConfiguration: proxy,
+            proxyConfiguration: requestProxyConfiguration,
             requestHandlerTimeoutSecs: CRAWLER_REQUEST_HANDLER_TIMEOUT_SECS,
+            useSessionPool: false,
             autoscaledPoolOptions: {
                 desiredConcurrency: input.desiredConcurrency,
             },