refactor!: Split off ConcurrencySystem from AutoscaledPool

[janbuchar]

• closes Accept a pre-built AutoscaledPool instance in crawler options #3886
• contributes to Minimize public interface of Crawlee components #3109

Accepting a whole pre-built AutoscaledPool (as the issue suggested) can't work — the pool has a single task source, so a shared instance ignores the second crawler's work. Split out the only shareable part instead: a new ConcurrencySystem (snapshotter + system status + budget + autoscaling) that AutoscaledPool delegates to. Inject the same one into several crawlers to cap their combined concurrency.

• Pool and crawlers depend on a minimal read-only IConcurrencySystem contract; task booking is atomic (including the per-minute cap), so sharing a budget can't oversubscribe it.
• Inject-or-default ownership (Unify the inject-or-default ownership pattern #3888): crawlers build and drive their own default (fresh per run()), while an injected system is caller-owned — start()/stop() it yourself, and run() throws if you forget.
• Runtime concurrency tuning moved from the pool to the ConcurrencySystem instance.
• Load-signal configuration is now a single loadSignals bag (per-resource tuning + custom), and a built-in signal can be switched off with false. Both evaluation windows apply to every signal alike, and LoadSignal.start() hands signals the window they'll be sampled over so retention isn't guesswork.
• Snapshotter/SystemStatus and the load-signal internals are now private to ConcurrencySystem: @crawlee/core drops 17 exported symbols and gains 11. Custom LoadSignals still supported.

Breaking changes are in the upgrading guide.

[barjin]

Thank you @janbuchar , a few initial ideas I got reading this ⬇️

[barjin]

Perhaps we should do this in the catch (or finally) branch as well?

import { BasicCrawler, CriticalError } from './packages/basic-crawler/src/index.js';

const c = new BasicCrawler({
    requestHandler: async () => {
        throw new CriticalError('Request failed');
    },
});

await c.run(['https://crawlee.dev/x']).catch(() => {}); // do not exit

console.log(c.autoscaledPool?.currentConcurrency);
// ⬆️  will log 1, as CriticalError goes straight to the AutoscaledPool, but the catch branch doesn't `registerTaskEnd`

Since this PR wants users to share the ConcurrencySystem, this might be a problem.

[janbuchar]

Nice catch! 3029248

[barjin]

We never check whether the ConcurrencySystem has actually been started. Since the .start() call is now users' responsibility, we should imo warn them, if they haven't called it.

migrating this

const c = new BasicCrawler({
    autoscaledPoolOptions: {
        maxConcurrency: 1,
    },
});

await c.run(['https://crawlee.dev/x']);

to this

const c = new BasicCrawler({
    concurrencySystem: new ConcurrencySystem({
        maxConcurrency: 1,
    }),
});

await c.run(['https://crawlee.dev/x']);

is easier than it seems (and currently logs no warning), and will leave the user with scaling hard-pinned to desiredConcurrency (iiuc).

[janbuchar]

True, I made it crash in that case. 640a297

[barjin]

where did the start() call go? 😄

[janbuchar]

640a297

[janbuchar]

@barjin this grew in scope by a fair bit, but I consider it done. Please re-check it.

[barjin]

Thank you @janbuchar , the changes are mostly fine with me 👍

My AI junior associate flagged the unfair scheduling with shared ConcurrencySystem. I managed to reproduce this and tried describing it in the comment.

On one hand, it's real and leads to consumer starvation, on the other, I cannot think of an easy fix. Let's perhaps make a separate issue and solve this in another PR?

[barjin]

This leads to unfair task allocation. If one consumer (A) has a head start, it:

1. A maxes out the shared ConcurrencySystem, e.g., running 10 tasks in parallel
2. B is introduced, wants to share the ConcurrencySystem
3. Every A task finished will immediately spawn another A task, leading to B starving.

Basically, the resource allocation is determined at initialization time, but won't change afterwards (until either consumer doesn't have any requests to process).

minimal example:

import { AutoscaledPool, ConcurrencySystem } from './packages/core/src/index.js';

const concurrencySystem = new ConcurrencySystem({
    minConcurrency: 10,
    maxConcurrency: 10,
    desiredConcurrency: 10,
    loggingIntervalSecs: null,
    loadSignals: { memory: false, eventLoop: false, cpu: false, client: false },
});

const done = { A: 0, B: 0 };
const pool = (name: 'A' | 'B') =>
    new AutoscaledPool({
        concurrencySystem,
        isFinishedFunction: async () => false,
        isTaskReadyFunction: async () => true,
        runTaskFunction: async () => {
            await new Promise((r) => setTimeout(r, 250));
            done[name]++;
        },
    });

const [a, b] = [pool('A'), pool('B')];
await concurrencySystem.start();

void a.run();
await new Promise((r) => setTimeout(r, 100)); // long enough for A to fill all 10 slots
void b.run();
await new Promise((r) => setTimeout(r, 2000));

console.log(done); // { A: 80, B: 0 }

await a.abort();
await b.abort();
await concurrencySystem.stop();

[janbuchar]

Yeah, I don't think we absolutely need to come up with a solution, but maybe we could make it possible for custom concurrency systems to resolve this? What if the interface accepted a crawler identity in the task allocation methods? 🤔

[janbuchar]

80e26e5