Skip to content

[pull] master from apify:master #294

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pull merged 2 commits into from
Sep 2, 2026
Merged

[pull] master from apify:master #294

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e9286be
chore(deps): dependency sweep — security bump + in-range freshening (…
apify-deps[bot] Sep 2, 2026
c3bd464
fix(basic): isolate crawler.run() from the caller's storage transacti…
barjin Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
"@opentelemetry/api": "^1.9.0",
"@opentelemetry/exporter-trace-otlp-grpc": "^0.210.0",
"@opentelemetry/resources": "^2.0.0",
"@opentelemetry/sdk-node": "^0.210.0",
"@opentelemetry/sdk-node": "^0.217.0",
"@opentelemetry/sdk-trace-base": "^2.0.0",
"@opentelemetry/semantic-conventions": "^1.39.0",
"apify": "*",
Expand Down
259 changes: 135 additions & 124 deletions packages/basic-crawler/src/internals/basic-crawler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ import type { ReadonlyDeep } from 'type-fest';
import { z } from 'zod';

import { LruCache } from '@apify/datastructures';
import { addTimeoutToPromise, extendTimeout, TimeoutError, tryCancel } from '@apify/timeout';
import { addTimeoutToPromise, extendTimeout, storage as timeoutStorage, TimeoutError, tryCancel } from '@apify/timeout';
import { cryptoRandomObjectId } from '@apify/utilities';

import {
Expand Down Expand Up @@ -1699,148 +1699,159 @@ export class BasicCrawler<
* @param [options] Options for the request queue.
*/
async run(requests?: TypedRequestsLike<Routes>, options?: CrawlerRunOptions): Promise<FinalStatistics> {
if (this.running) {
throw new Error(
'This crawler instance is already running, you can add more requests to it via `crawler.addRequests()`.',
);
}

const { purgeRequestQueue, ...addRequestsOptions } = options ?? {};

if (this.hasFinishedBefore) {
// When executing the run method for the second time explicitly,
// we need to purge the RQ to allow processing the same requests again — this is important so users can
// pass in failed requests back to the `crawler.run()`, otherwise they would be considered as handled and
// ignored — as a failed request is still handled.
// `purgeRequestQueue` unset purges only storage the crawler opened itself (see `#purgeableExtent`);
// `true` also purges a caller-supplied manager, `false` purges nothing.
if (purgeRequestQueue === undefined && this.#purgeableExtent === 'ambiguous') {
throw new Error(
'Cannot decide what to purge before running again: `sameDomainDelaySecs` paces the request ' +
'manager you supplied, so the per-domain queues that have to be emptied are the ' +
"crawler's while the manager underneath them is yours. Say which you want: " +
'`run(requests, { purgeRequestQueue: true })` empties both, `false` empties neither.',
);
}

if (purgeRequestQueue !== false && (this.#purgeableExtent === 'all' || purgeRequestQueue === true)) {
// One call from the outside in reaches everything the manager wraps, a pacer's per-domain queues
// included.
await this.requestManager?.purge?.();
}
// A crawl is the top level of its own transaction and timeout scope, not a participant in the caller's.
return withDirectStorageAccess(async () =>
timeoutStorage.exit(async () => {
if (this.running) {
throw new Error(
'This crawler instance is already running, you can add more requests to it via `crawler.addRequests()`.',
);
}

// A supplied statistics instance keeps whatever state it was handed - only wipe a default we built.
await this.#statisticsDep.ifOwned(async (stats) => {
stats.reset();
await stats.resetStore();
});
await this.#sessionPoolDep.ifOwned((pool) => pool.resetStore());
}
const { purgeRequestQueue, ...addRequestsOptions } = options ?? {};

if (this.hasFinishedBefore) {
// When executing the run method for the second time explicitly,
// we need to purge the RQ to allow processing the same requests again — this is important so users can
// pass in failed requests back to the `crawler.run()`, otherwise they would be considered as handled and
// ignored — as a failed request is still handled.
// `purgeRequestQueue` unset purges only storage the crawler opened itself (see `#purgeableExtent`);
// `true` also purges a caller-supplied manager, `false` purges nothing.
if (purgeRequestQueue === undefined && this.#purgeableExtent === 'ambiguous') {
throw new Error(
'Cannot decide what to purge before running again: `sameDomainDelaySecs` paces the request ' +
'manager you supplied, so the per-domain queues that have to be emptied are the ' +
"crawler's while the manager underneath them is yours. Say which you want: " +
'`run(requests, { purgeRequestQueue: true })` empties both, `false` empties neither.',
);
}

this.#unexpectedStop = false;
this.running = true;
this.#loggedPerRun.clear();
if (
purgeRequestQueue !== false &&
(this.#purgeableExtent === 'all' || purgeRequestQueue === true)
) {
// One call from the outside in reaches everything the manager wraps, a pacer's per-domain queues
// included.
await this.requestManager?.purge?.();
}

await purgeDefaultStorages({
onlyPurgeOnce: true,
storageBackend: serviceLocator.getStorageBackend(),
configuration: serviceLocator.getConfiguration(),
});
// A supplied statistics instance keeps whatever state it was handed - only wipe a default we built.
await this.#statisticsDep.ifOwned(async (stats) => {
stats.reset();
await stats.resetStore();
});
await this.#sessionPoolDep.ifOwned((pool) => pool.resetStore());
}

if (requests) {
await this.addRequests(requests, addRequestsOptions);
}
this.#unexpectedStop = false;
this.running = true;
this.#loggedPerRun.clear();

try {
await this.init();
await this.statistics.startCapturing();
} catch (error) {
// Clean up here before propagating, otherwise a failed startup would leave the process hanging.
await this.teardown().catch((teardownError) => {
this.log.exception(teardownError as Error, 'Cleaning up after a failed crawler startup failed.');
});
await purgeDefaultStorages({
onlyPurgeOnce: true,
storageBackend: serviceLocator.getStorageBackend(),
configuration: serviceLocator.getConfiguration(),
});

// The run never began, so let the instance be run again instead of leaving it wedged as `running`.
this.running = false;
throw error;
}
if (requests) {
await this.addRequests(requests, addRequestsOptions);
}

const periodicLogger = this.getPeriodicLogger();
this.setStatusMessage('Starting the crawler.', { level: 'INFO' });
try {
await this.init();
await this.statistics.startCapturing();
} catch (error) {
// Clean up here before propagating, otherwise a failed startup would leave the process hanging.
await this.teardown().catch((teardownError) => {
this.log.exception(
teardownError as Error,
'Cleaning up after a failed crawler startup failed.',
);
});

const sigintHandler = async () => {
this.log.warning(
'Pausing... Press CTRL+C again to force exit. To resume, do: CRAWLEE_PURGE_ON_START=0 npm start',
);
await this.pauseOnMigration();
await this.#autoscaledPool!.abort();
};
// The run never began, so let the instance be run again instead of leaving it wedged as `running`.
this.running = false;
throw error;
}

// Attach a listener to handle migration and aborting events gracefully.
const boundPauseOnMigration = this.pauseOnMigration.bind(this);
process.once('SIGINT', sigintHandler);
const eventManager = serviceLocator.getEventManager();
eventManager.on(EventType.MIGRATING, boundPauseOnMigration);
eventManager.on(EventType.ABORTING, boundPauseOnMigration);
const periodicLogger = this.getPeriodicLogger();
this.setStatusMessage('Starting the crawler.', { level: 'INFO' });

let stats = {} as FinalStatistics;
const sigintHandler = async () => {
this.log.warning(
'Pausing... Press CTRL+C again to force exit. To resume, do: CRAWLEE_PURGE_ON_START=0 npm start',
);
await this.pauseOnMigration();
await this.#autoscaledPool!.abort();
};

try {
await this.#autoscaledPool!.run();
} finally {
await this.statistics.stopCapturing();
await this.teardown();

process.off('SIGINT', sigintHandler);
eventManager.off(EventType.MIGRATING, boundPauseOnMigration);
eventManager.off(EventType.ABORTING, boundPauseOnMigration);

const finalStats = this.statistics.calculate();
stats = {
requestsFinished: this.statistics.state.requestsFinished,
requestsFailed: this.statistics.state.requestsFailed,
retryHistogram: this.statistics.requestRetryHistogram,
...finalStats,
};
this.log.info('Final request statistics:', stats as unknown as Record<string, unknown>);
// Attach a listener to handle migration and aborting events gracefully.
const boundPauseOnMigration = this.pauseOnMigration.bind(this);
process.once('SIGINT', sigintHandler);
const eventManager = serviceLocator.getEventManager();
eventManager.on(EventType.MIGRATING, boundPauseOnMigration);
eventManager.on(EventType.ABORTING, boundPauseOnMigration);

if (this.statistics.errorTracker.total !== 0) {
const prettify = ([count, info]: [number, string[]]) =>
`${count}x: ${info.at(-1)!.trim()} (${info[0]})`;
let stats = {} as FinalStatistics;

this.log.info(`Error analysis:`, {
totalErrors: this.statistics.errorTracker.total,
uniqueErrors: this.statistics.errorTracker.getUniqueErrorCount(),
mostCommonErrors: this.statistics.errorTracker.getMostPopularErrors(3).map(prettify),
});
}
try {
await this.#autoscaledPool!.run();
} finally {
await this.statistics.stopCapturing();
await this.teardown();

process.off('SIGINT', sigintHandler);
eventManager.off(EventType.MIGRATING, boundPauseOnMigration);
eventManager.off(EventType.ABORTING, boundPauseOnMigration);

const finalStats = this.statistics.calculate();
stats = {
requestsFinished: this.statistics.state.requestsFinished,
requestsFailed: this.statistics.state.requestsFailed,
retryHistogram: this.statistics.requestRetryHistogram,
...finalStats,
};
this.log.info('Final request statistics:', stats as unknown as Record<string, unknown>);

if (this.statistics.errorTracker.total !== 0) {
const prettify = ([count, info]: [number, string[]]) =>
`${count}x: ${info.at(-1)!.trim()} (${info[0]})`;

this.log.info(`Error analysis:`, {
totalErrors: this.statistics.errorTracker.total,
uniqueErrors: this.statistics.errorTracker.getUniqueErrorCount(),
mostCommonErrors: this.statistics.errorTracker.getMostPopularErrors(3).map(prettify),
});
}

const client = serviceLocator.getStorageBackend();
const client = serviceLocator.getStorageBackend();

if (client.teardown) {
let finished = false;
setTimeout(() => {
if (!finished) {
this.log.info('Waiting for the storage to write its state to file system.');
if (client.teardown) {
let finished = false;
setTimeout(() => {
if (!finished) {
this.log.info('Waiting for the storage to write its state to file system.');
}
}, 1000);
await client.teardown();
finished = true;
}
}, 1000);
await client.teardown();
finished = true;
}

periodicLogger.stop();
this.setStatusMessage(
`Finished! Total ${this.statistics.state.requestsFinished + this.statistics.state.requestsFailed} requests: ${
this.statistics.state.requestsFinished
} succeeded, ${this.statistics.state.requestsFailed} failed.`,
{ isStatusMessageTerminal: true, level: 'INFO' },
);
periodicLogger.stop();
this.setStatusMessage(
`Finished! Total ${this.statistics.state.requestsFinished + this.statistics.state.requestsFailed} requests: ${
this.statistics.state.requestsFinished
} succeeded, ${this.statistics.state.requestsFailed} failed.`,
{ isStatusMessageTerminal: true, level: 'INFO' },
);

this.running = false;
this.hasFinishedBefore = true;
}
this.running = false;
this.hasFinishedBefore = true;
}

return stats;
return stats;
}),
);
}

/**
Expand Down
Loading
Loading