Check if ConcurrencySystem is running before starting a crawl

docs/public-api/crawlee-core.api.md

@@ -184,6 +184,7 @@ export class ConcurrencySystem implements IConcurrencySystem {
     set desiredConcurrency(value: number);
     getCurrentStatus(): SystemInfo;
     hasCapacityForTask(): boolean;
+    get isRunning(): boolean;
     get maxConcurrency(): number;
     set maxConcurrency(value: number);
     get minConcurrency(): number;
@@ -706,6 +707,7 @@ export interface IConcurrencySystem {
     readonly currentConcurrency: number;
     readonly desiredConcurrency: number;
     hasCapacityForTask(): boolean;
+    readonly isRunning?: boolean;
     registerTaskEnd(): void;
     tryRegisterTaskStart(): boolean;
 }

docs/upgrading/upgrading_v4.md

@@ -1177,7 +1177,7 @@ Both the pool and the crawlers depend only on the `IConcurrencySystem` interface
 
 All scaling/snapshotter options were **removed** from `AutoscaledPoolOptions` and now live on `ConcurrencySystemOptions`: `minConcurrency`, `maxConcurrency`, `desiredConcurrency`, `desiredConcurrencyRatio`, `scaleUpStepRatio`, `scaleDownStepRatio`, `loggingIntervalSecs`, `autoscaleIntervalSecs`, `maxTasksPerMinute`, and `snapshotterOptions`. The `systemStatusOptions` bag is gone entirely — its two useful members were hoisted onto `ConcurrencySystemOptions` directly as `loadSignals` (custom load signals) and `currentHistorySecs`. In place of all this, `AutoscaledPoolOptions` gained a **required** `concurrencySystem`.
 
-`AutoscaledPool` also no longer manages the system's lifecycle: it assumes the `ConcurrencySystem` is already running. Whoever owns the system must `start()` it before `pool.run()` and `stop()` it afterwards. (Within Crawlee this is handled by the crawler for the pool it builds; you only need to care about this if you drive an `AutoscaledPool` directly.)
+`AutoscaledPool` also no longer manages the system's lifecycle: it requires the `ConcurrencySystem` to be running already. Whoever owns the system must `start()` it before `pool.run()` and `stop()` it afterwards — `pool.run()` throws if the system it borrows was never started, instead of running a whole crawl with load monitoring and autoscaling switched off. (Within Crawlee this is handled by the crawler for the pool it builds; you only need to care about this if you inject a system or drive an `AutoscaledPool` directly.)
 
 One subtle behavioral consequence: pausing no longer suspends autoscaling. Previously, `pool.pause()` also stopped the pool's autoscaling loop ("don't scale if paused"). The autoscaling interval now lives on the `ConcurrencySystem`, which has no knowledge of its borrowing pools' pause state — deliberately so, since with a shared system other pools may still be running and need scaling. As a result, a running system keeps evaluating (and possibly scaling down) the desired concurrency and keeps emitting its periodic state log even while every pool that consults it is paused. Scaling *up* remains effectively blocked while paused, because the current concurrency drains below the required ratio of the desired concurrency. If you pause for extended periods and want the system fully quiet, `stop()` the system (if you own it) and `start()` it again before resuming.
 
@@ -1251,16 +1251,24 @@ const crawler = new CheerioCrawler({
 ```typescript
 import { ConcurrencySystem } from '@crawlee/core';
 
-const crawler = new CheerioCrawler({
-    concurrencySystem: new ConcurrencySystem({
-        desiredConcurrency: 10,
-        maxTasksPerMinute: 120,
-        currentHistorySecs: 10,
-    }),
-    requestHandler,
+const concurrencySystem = new ConcurrencySystem({
+    desiredConcurrency: 10,
+    maxTasksPerMinute: 120,
+    currentHistorySecs: 10,
 });
+await concurrencySystem.start();
+
+const crawler = new CheerioCrawler({ concurrencySystem, requestHandler });
+
+try {
+    await crawler.run();
+} finally {
+    await concurrencySystem.stop();
+}
 ```
 
+Note that an injected system is **yours to run**: the crawler never starts or stops a system it did not build, since it cannot know when the last crawler sharing it is done. A system you forget to `start()` would monitor no load and never rescale, so rather than silently running at a fixed concurrency, `run()` **throws** — check `isRunning` if you need to know whether some other owner already started a system handed to you.
+
 For the common case, the shortcuts are enough and need no `ConcurrencySystem`:
 
 ```typescript

packages/core/src/autoscaling/autoscaled_pool.ts

@@ -228,8 +228,24 @@ export class AutoscaledPool {
     /**
      * Runs the auto-scaled pool. Returns a promise that gets resolved or rejected once
      * all the tasks are finished or one of them fails.
+     *
+     * Throws if the {@apilink IConcurrencySystem|concurrency system} it borrows was never started — the pool assumes
+     * a running governor and cannot start one it does not own.
      */
     async run(): Promise<void> {
+        // Fail fast rather than run against a dead governor: without a running snapshotter and autoscaling interval
+        // the desired concurrency never moves, so the whole run would silently proceed at a fixed concurrency. This
+        // has to be checked here, on an awaited path — the capacity queries inside the task loop are driven by
+        // intervals and `setImmediate`, where a throw would become an unhandled rejection and hang `run()` forever.
+        if (this.concurrencySystem.isRunning === false) {
+            throw new CriticalError(
+                'The ConcurrencySystem this AutoscaledPool borrows has not been started, so system load would not be ' +
+                    'monitored and the concurrency would never be adjusted. Whoever creates a ConcurrencySystem owns ' +
+                    'its lifecycle: call `await concurrencySystem.start()` before running the pools or crawlers that ' +
+                    'use it, and `await concurrencySystem.stop()` once they are all done.',
+            );
+        }
+
         const poolPromise = new Promise((resolve, reject) => {
             this.resolve = resolve;
             this.reject = reject;

packages/core/src/autoscaling/concurrency_system.ts

@@ -125,6 +125,15 @@ export interface IConcurrencySystem {
      */
     readonly currentConcurrency: number;
 
+    /**
+     * Whether a governor that needs starting has actually been started. {@apilink AutoscaledPool.run|`pool.run()`}
+     * refuses to run against a governor that reports `false`, since a governor nobody started monitors nothing and
+     * never rescales — the run would silently proceed at a fixed concurrency. Omit this member entirely (leaving it
+     * `undefined`) if the implementation has no startup lifecycle and is always ready; only an explicit `false` is
+     * treated as an error.
+     */
+    readonly isRunning?: boolean;
+
     /**
      * The core, shareable decision: may **one more** task start right now? A cheap pre-check the pool consults
      * before querying task readiness; the actual booking happens through
@@ -323,6 +332,14 @@ export class ConcurrencySystem implements IConcurrencySystem {
         return this._currentConcurrency;
     }
 
+    /**
+     * Whether the system is currently started, i.e. monitoring load and autoscaling the budget. Borrowing pools read
+     * this to refuse to run against a system nobody started — see {@apilink IConcurrencySystem.isRunning}.
+     */
+    get isRunning(): boolean {
+        return this.running;
+    }
+
     /**
      * Boots the underlying snapshotter and the autoscaling interval. Idempotent — calling it on an
      * already-running system is a no-op, so a shared system started by its owner isn't restarted when handed to

test/core/autoscaling/concurrency_system.test.ts

@@ -95,6 +95,74 @@ describe('ConcurrencySystem', () => {
             await system.stop();
             expect(stopSpy).not.toHaveBeenCalled();
         });
+
+        test('isRunning reflects the lifecycle', async () => {
+            const system = new ConcurrencySystem();
+            expect(system.isRunning).toBe(false);
+
+            await system.start();
+            expect(system.isRunning).toBe(true);
+
+            await system.stop();
+            expect(system.isRunning).toBe(false);
+        });
+
+        test('a pool refuses to run against a system that was never started', async () => {
+            const system = new ConcurrencySystem();
+            const runTaskFunction = vitest.fn(async () => {});
+
+            const pool = new AutoscaledPool({
+                concurrencySystem: system,
+                runTaskFunction,
+                isFinishedFunction: async () => false,
+                isTaskReadyFunction: async () => true,
+            });
+
+            // Forgetting to start an injected system would pin concurrency for the whole run, so fail loudly
+            // before any work happens instead of silently misbehaving.
+            await expect(pool.run()).rejects.toThrow(/has not been started/);
+            expect(runTaskFunction).not.toHaveBeenCalled();
+        });
+
+        test('a pool runs happily against a started system', async () => {
+            const system = new ConcurrencySystem();
+            await system.start();
+
+            let done = 0;
+            const pool = new AutoscaledPool({
+                concurrencySystem: system,
+                runTaskFunction: async () => {
+                    done++;
+                },
+                isFinishedFunction: async () => done > 0,
+                isTaskReadyFunction: async () => done === 0,
+            });
+
+            await expect(pool.run()).resolves.toBeUndefined();
+            await system.stop();
+        });
+
+        test('a governor with no startup lifecycle is accepted', async () => {
+            // `isRunning` is optional in the IConcurrencySystem contract - a governor that needs no starting simply
+            // omits it, and only an explicit `false` is treated as an error.
+            let done = 0;
+            const pool = new AutoscaledPool({
+                concurrencySystem: {
+                    desiredConcurrency: 1,
+                    currentConcurrency: 0,
+                    hasCapacityForTask: () => true,
+                    tryRegisterTaskStart: () => true,
+                    registerTaskEnd: () => {},
+                },
+                runTaskFunction: async () => {
+                    done++;
+                },
+                isFinishedFunction: async () => done > 0,
+                isTaskReadyFunction: async () => done === 0,
+            });
+
+            await expect(pool.run()).resolves.toBeUndefined();
+        });
     });
 
     describe('shared across pools', () => {

test/core/crawlers/basic_crawler.test.ts

@@ -531,7 +531,10 @@ describe('BasicCrawler', () => {
             maxRequestsPerMinute: 789,
         });
 
+        // An injected system is ours to run - the crawler refuses to run against one that was never started.
+        await injectedSystem.start();
         await Promise.all([shortcuts.run(), injected.run()]);
+        await injectedSystem.stop();
 
         expect(collect(shortcuts)).toEqual({ minConcurrency: 123, maxConcurrency: 456, maxTasksPerMinute: 789 });
         expect(collect(injected)).toEqual({ minConcurrency: 16, maxConcurrency: 32, maxTasksPerMinute: 64 });