From c997000f03595617cf1e4dc9ed841f41a93efa31 Mon Sep 17 00:00:00 2001 From: nikitachapovskii-dev Date: Wed, 7 Oct 2026 16:20:31 +0200 Subject: [PATCH 1/5] fix: set the log level per run and answer callers when the run migrates --- src/input.ts | 2 -- src/main.ts | 10 ++++++---- src/responses.ts | 21 ++++++++------------- tests/input.test.ts | 20 +++++++++++++++++++- tests/responses.test.ts | 36 ++++++++++++++++++++++++++++++++++++ 5 files changed, 69 insertions(+), 20 deletions(-) create mode 100644 tests/responses.test.ts diff --git a/src/input.ts b/src/input.ts index 7d71bd1..bd59ac4 100644 --- a/src/input.ts +++ b/src/input.ts @@ -129,8 +129,6 @@ async function processInputInternal( removeCookieWarnings, } = input; - log.setLevel(debugMode ? log.LEVELS.DEBUG : log.LEVELS.INFO); - const contentScraperSettings: ContentScraperSettings = { debugMode, dynamicContentWaitSecs, diff --git a/src/main.ts b/src/main.ts index 204ff36..e16a77e 100644 --- a/src/main.ts +++ b/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,14 @@ import { isActorStandby } from './utils.js'; await Actor.init(); -Actor.on('migrating', () => { - addTimeoutToAllResponses(60); -}); +Actor.on('migrating', failAllResponsesOnMigration); const originalInput = await Actor.getInput>() ?? {} 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. +log.setLevel(originalInput.debugMode ? log.LEVELS.DEBUG : log.LEVELS.INFO); + if (isActorStandby()) { log.info('Actor is running in the STANDBY mode.'); diff --git a/src/responses.ts b/src/responses.ts index f9dae1e..9605545 100644 --- a/src/responses.ts +++ b/src/responses.ts @@ -163,19 +163,14 @@ export function sendResponseIfFinished(responseId: string) { sendResponseOk(responseId, sortResultsByRank(res)); } } + /** - * Add timeout to all responses when actor is migrating (source: SuperScraper). + * 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. */ -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.'); } -}; +} diff --git a/tests/input.test.ts b/tests/input.test.ts index 904d79f..ef0b241 100644 --- a/tests/input.test.ts +++ b/tests/input.test.ts @@ -1,7 +1,10 @@ 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'; describe('warnIfLowMemoryForPlaywright', () => { afterEach(() => { @@ -32,3 +35,18 @@ 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 () => { + // 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}`)); + + expect(log.getLevel()).toBe(log.LEVELS.INFO); + }); +}); diff --git a/tests/responses.test.ts b/tests/responses.test.ts new file mode 100644 index 0000000..9d97348 --- /dev/null +++ b/tests/responses.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from 'vitest'; + +import { ContentCrawlerStatus } from '../src/const.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 the moment the migration is announced, instead of giving + // the crawl a grace period the process may not live long enough to honour. + 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 } }]); + }); +}); From 366b9432b41d231d265228ad79887023046c0d8f Mon Sep 17 00:00:00 2001 From: nikitachapovskii-dev Date: Thu, 8 Oct 2026 11:05:53 +0200 Subject: [PATCH 2/5] fix: size the shared crawler pool from the run, not from a request --- src/input.ts | 8 ++++++++ src/main.ts | 3 +-- src/responses.ts | 5 ++--- tests/input.test.ts | 30 ++++++++++++++++++++++++------ tests/responses.test.ts | 3 +-- 5 files changed, 36 insertions(+), 13 deletions(-) diff --git a/src/input.ts b/src/input.ts index bd59ac4..2a2c41c 100644 --- a/src/input.ts +++ b/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) { 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) { // 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); diff --git a/src/main.ts b/src/main.ts index e16a77e..d6f79fa 100644 --- a/src/main.ts +++ b/src/main.ts @@ -17,8 +17,7 @@ Actor.on('migrating', failAllResponsesOnMigration); const originalInput = await Actor.getInput>() ?? {} 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()) { diff --git a/src/responses.ts b/src/responses.ts index 9605545..92d36e4 100644 --- a/src/responses.ts +++ b/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()]) { diff --git a/tests/input.test.ts b/tests/input.test.ts index ef0b241..0731ecc 100644 --- a/tests/input.test.ts +++ b/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); + }); +}); diff --git a/tests/responses.test.ts b/tests/responses.test.ts index 9d97348..c30edeb 100644 --- a/tests/responses.test.ts +++ b/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'; From 1e19af89bb36cb2d959212ac3f567352b8ed61a2 Mon Sep 17 00:00:00 2001 From: nikitachapovskii-dev Date: Thu, 8 Oct 2026 15:39:29 +0200 Subject: [PATCH 3/5] fix: tell the caller when the search fails, and stop writing results nobody awaits --- src/crawlers.ts | 3 +-- src/request-handler.ts | 2 ++ tests/responses.test.ts | 19 +++++++++++++++++++ 3 files changed, 22 insertions(+), 2 deletions(-) diff --git a/src/crawlers.ts b/src/crawlers.ts index a05bc71..8fadab9 100644 --- a/src/crawlers.ts +++ b/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); }, })); diff --git a/src/request-handler.ts b/src/request-handler.ts index 6df0ad8..a324241 100644 --- a/src/request-handler.ts +++ b/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()); diff --git a/tests/responses.test.ts b/tests/responses.test.ts index c30edeb..0bf84e7 100644 --- a/tests/responses.test.ts +++ b/tests/responses.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import { ContentCrawlerStatus } from '../src/const.js'; +import { createAndStartSearchCrawler } from '../src/crawlers.js'; import { addEmptyResultToResponse, addResultToResponse, @@ -33,3 +34,21 @@ describe('a run that migrates to another server', () => { 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; + }; + + 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'); + }); +}); From f9cc5bb5729a66f59702eb5fd12644a9802f46de Mon Sep 17 00:00:00 2001 From: nikitachapovskii-dev Date: Mon, 12 Oct 2026 11:00:53 +0200 Subject: [PATCH 4/5] fix: let in-flight requests finish before the run moves, and refuse new ones --- src/main.ts | 4 ++-- src/responses.ts | 38 +++++++++++++++++++++++++----- src/search.ts | 5 +++- tests/responses.test.ts | 52 ++++++++++++++++++++++++----------------- 4 files changed, 68 insertions(+), 31 deletions(-) diff --git a/src/main.ts b/src/main.ts index d6f79fa..2d8291c 100644 --- a/src/main.ts +++ b/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 { failAllResponsesOnMigration } from './responses.js'; +import { startMigration } from './responses.js'; import { handleSearchNormalMode } from './search.js'; import { createServer } from './server.js'; import { findActorTip, storeActorTip } from './tips.js'; @@ -13,7 +13,7 @@ import { isActorStandby } from './utils.js'; await Actor.init(); -Actor.on('migrating', failAllResponsesOnMigration); +Actor.on('migrating', async () => startMigration()); const originalInput = await Actor.getInput>() ?? {} as Input; diff --git a/src/responses.ts b/src/responses.ts index 92d36e4..dc2025e 100644 --- a/src/responses.ts +++ b/src/responses.ts @@ -1,5 +1,6 @@ import { log } from 'apify'; import type { RequestOptions } from 'crawlee'; +import { sleep } from 'crawlee'; import { ContentCrawlerStatus } from './const.js'; import type { ContentCrawlerUserData, Output } from './types.js'; @@ -164,12 +165,37 @@ export function sendResponseIfFinished(responseId: string) { } } +export const MIGRATION_MESSAGE = 'Actor had to migrate to another server. Please, retry your request.'; + /** - * 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. + * Crawlee already stops taking new pages on migration and waits up to `SAFE_MIGRATION_WAIT_MILLIS`, + * 20 seconds, for the ones in flight; `Actor.reboot()` waits for that listener. Answering sooner + * would throw away work the crawler is still finishing, so wait a little longer than it does. */ -export function failAllResponsesOnMigration() { - for (const responseId of [...responseData.keys()]) { - sendResponseError(responseId, 'Actor had to migrate to another server. Please, retry your request.'); - } +const MIGRATION_GRACE_MILLIS = 25_000; + +let migration: Promise | undefined; + +/** Whether the platform has announced that this run is moving to another server. */ +export function isMigrating() { + return migration !== undefined; +} + +/** + * Lets the requests already in flight finish, then answers whoever is left. Returned so that the + * caller can hold the move back, and kept so that the second announcement joins the first. + */ +export async function startMigration(graceMillis = MIGRATION_GRACE_MILLIS) { + migration ??= (async () => { + const deadline = Date.now() + graceMillis; + while (responseData.size > 0 && Date.now() < deadline) { + await sleep(200); + } + + for (const responseId of [...responseData.keys()]) { + sendResponseError(responseId, MIGRATION_MESSAGE); + } + })(); + + return migration; } diff --git a/src/search.ts b/src/search.ts index 82289a3..0593d55 100644 --- a/src/search.ts +++ b/src/search.ts @@ -9,7 +9,7 @@ import { addContentCrawlRequest, addSearchRequest, createAndStartContentCrawler, import { UserInputError } from './errors.js'; import { processInput } from './input.js'; import { getMiniActor } from './mini-actors.js'; -import { createResponsePromise } from './responses.js'; +import { createResponsePromise, isMigrating, MIGRATION_MESSAGE } from './responses.js'; import type { ContentCrawlerOptions, ContentScraperSettings, @@ -107,6 +107,9 @@ function prepareRequest( * Returns a promise that resolves to the final results array of Output objects. */ async function runSearchProcess(params: Partial, actorRequestId?: string): Promise { + // Starting a crawl the move is about to interrupt only costs the caller their own timeout. + if (isMigrating()) throw new Error(MIGRATION_MESSAGE); + // Process the query parameters the same way as normal inputs const { input, diff --git a/tests/responses.test.ts b/tests/responses.test.ts index 0bf84e7..32254c9 100644 --- a/tests/responses.test.ts +++ b/tests/responses.test.ts @@ -6,32 +6,40 @@ import { addEmptyResultToResponse, addResultToResponse, createResponsePromise, - failAllResponsesOnMigration, + sendResponseIfFinished, + startMigration, } 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 handled = (responseId: string, url: string) => { + addEmptyResultToResponse(responseId, { url, uniqueKey: url } as never); + addResultToResponse(responseId, url, { + crawl: { requestStatus: ContentCrawlerStatus.HANDLED }, + metadata: { url }, + searchResult: { rank: 1 }, + } as never); +}; + +// A run moves only once, so every outcome of that one move belongs in a single test. +describe('a run that moves to another server', () => { + it('lets callers finish, keeps what the rest already have, and tells the empty ones to retry', async () => { + const inTime = createResponsePromise('in-time', 300); + const partial = createResponsePromise('partial', 300); + const empty = createResponsePromise('empty', 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 } }]); + const moved = startMigration(500); + handled('in-time', url); + sendResponseIfFinished('in-time'); + handled('partial', url); + addEmptyResultToResponse('partial', { url: 'https://example.com/pending', uniqueKey: 'pending', userData: { searchResult: { rank: 2 } } } as never); + await moved; + + await expect(inTime).resolves.toMatchObject([{ metadata: { url } }]); + await expect(partial).resolves.toMatchObject([ + { metadata: { url } }, + { crawl: { requestStatus: ContentCrawlerStatus.FAILED } }, + ]); + await expect(empty).rejects.toThrow('Actor had to migrate to another server. Please, retry your request.'); }); }); From 0a08078a0f07c65170996f12408d4ce566410ffb Mon Sep 17 00:00:00 2001 From: nikitachapovskii-dev Date: Mon, 12 Oct 2026 11:16:15 +0200 Subject: [PATCH 5/5] fix: stop charging for a search whose caller is already gone --- src/crawlers.ts | 14 +++++++++++++- src/input.ts | 3 +-- src/responses.ts | 5 ++--- tests/charging.standby.test.ts | 14 +++++++++++++- tests/input.test.ts | 5 +++-- 5 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/crawlers.ts b/src/crawlers.ts index 8fadab9..e2ab0e5 100644 --- a/src/crawlers.ts +++ b/src/crawlers.ts @@ -27,7 +27,13 @@ import { requestHandlerPlaywright, TEXT_DOCUMENT_CONTENT_TYPES, } from './request-handler.js'; -import { addEmptyResultToResponse, addResultToResponse, sendResponseError, sendResponseIfFinished } from './responses.js'; +import { + addEmptyResultToResponse, + addResultToResponse, + responseData, + sendResponseError, + sendResponseIfFinished, +} from './responses.js'; import type { ContentCrawlerOptions, ContentCrawlerUserData, @@ -157,11 +163,17 @@ export async function createAndStartSearchCrawler( // refuses stays free - but before anything is enqueued, or the response could beat the charge // and invalidate the caller's request ID. The flag rides along with the request, so a retry // of this handler cannot charge the query twice. + // The caller can be gone by now: timed out, failed, or the run is moving. + if (!responseData.has(request.userData.responseId)) return; + if (organicResults.length > 0 && !request.userData.isSearchChargeAttempted) { request.userData.isSearchChargeAttempted = true; await chargeSearch({ actorRequestId, idempotencyKey: request.uniqueKey }); } + // The charge is awaited, so they can have gone during it. + if (!responseData.has(request.userData.responseId)) return; + // Merge with previously collected results and deduplicate const allResults = [...collectedResults, ...organicResults]; const deduplicated = deduplicateResults(allResults); diff --git a/src/input.ts b/src/input.ts index 2a2c41c..2915f32 100644 --- a/src/input.ts +++ b/src/input.ts @@ -97,8 +97,7 @@ export async function processInput(originalInput: Partial) { // 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. + // A request builds these options when no crawler is cached yet, and must not size the shared pool. if (isActorStandby()) input.desiredConcurrency = runDesiredConcurrency; const contentCrawlerOptions: ContentCrawlerOptions = input.scrapingTool === 'raw-http' diff --git a/src/responses.ts b/src/responses.ts index dc2025e..58c25c7 100644 --- a/src/responses.ts +++ b/src/responses.ts @@ -176,14 +176,13 @@ const MIGRATION_GRACE_MILLIS = 25_000; let migration: Promise | undefined; -/** Whether the platform has announced that this run is moving to another server. */ export function isMigrating() { return migration !== undefined; } /** - * Lets the requests already in flight finish, then answers whoever is left. Returned so that the - * caller can hold the move back, and kept so that the second announcement joins the first. + * Returned so the caller can hold the move back, and kept so the second announcement joins the first + * rather than starting a second wait. */ export async function startMigration(graceMillis = MIGRATION_GRACE_MILLIS) { migration ??= (async () => { diff --git a/tests/charging.standby.test.ts b/tests/charging.standby.test.ts index f829474..f355b18 100644 --- a/tests/charging.standby.test.ts +++ b/tests/charging.standby.test.ts @@ -207,7 +207,7 @@ describe('Charging from standby requests', () => { * Submits a search request straight to the search crawler, pointed at a stand-in result page on the * test server so that the crawler can be exercised without reaching Google. */ - async function submitSearch(path: string, userData: Partial = {}) { + async function submitSearch(path: string, userData: Partial = {}, { waiting = true } = {}) { const request = createSearchRequest({ query: 'hello world', maxResults: 1, @@ -227,6 +227,8 @@ describe('Charging from standby requests', () => { ...userData, }); request.url = `${baseUrl}${path}`; + // A search always has a caller waiting on it, unless that caller has gone. + if (waiting) createResponsePromise(request.userData!.responseId, 30).catch(() => undefined); await addSearchRequest(request, searchCrawlerKey); return request.uniqueKey!; } @@ -239,6 +241,16 @@ describe('Charging from standby requests', () => { expect(charging.chargeSearch).toHaveBeenCalledExactlyOnceWith({ actorRequestId: 'request123', idempotencyKey: uniqueKey }); }); + it('does not charge for a search whose caller has already gone', async () => { + const finishedBefore = searchCrawler!.stats.state.requestsFinished; + + await submitSearch('/serp', {}, { waiting: false }); + + await vi.waitFor(() => expect(searchCrawler!.stats.state.requestsFinished).toBe(finishedBefore + 1), { timeout: 10_000 }); + expect(charging.chargeSearch).not.toHaveBeenCalled(); + expect(charging.chargeFetch).not.toHaveBeenCalled(); + }); + it('does not charge for a search that Google refuses with an empty page', async () => { const finishedBefore = searchCrawler!.stats.state.requestsFinished; diff --git a/tests/input.test.ts b/tests/input.test.ts index 0731ecc..ef5c27e 100644 --- a/tests/input.test.ts +++ b/tests/input.test.ts @@ -2,7 +2,7 @@ import { Actor } from 'apify'; import { log } from 'crawlee'; import { afterEach, describe, expect, it, vi } from 'vitest'; -import { processInput, warnIfLowMemoryForPlaywright } from '../src/input.js'; +import { processInput, processStandbyInput, warnIfLowMemoryForPlaywright } from '../src/input.js'; import { parseParameters } from '../src/utils.js'; process.env.ACTOR_FULL_NAME = 'apify/rag-web-browser'; @@ -60,11 +60,12 @@ describe('the shared crawler pool', () => { it('is sized by the run, not by whichever caller happens to build it', async () => { vi.spyOn(Actor, 'getEnv').mockReturnValue({ metaOrigin: 'STANDBY' } as never); + await processStandbyInput(parseParameters(`?query=hello&desiredConcurrency=23&proxyConfiguration=${CALLER_PROXY}`)); const { contentCrawlerOptions } = await processInput( parseParameters(`?query=hello&desiredConcurrency=17&proxyConfiguration=${CALLER_PROXY}`), ); - expect(contentCrawlerOptions.crawlerOptions.autoscaledPoolOptions?.desiredConcurrency).toBe(5); + expect(contentCrawlerOptions.crawlerOptions.autoscaledPoolOptions?.desiredConcurrency).toBe(23); }); });