Tell load signals the window they will be sampled over

Snapshot retention was out-of-band knowledge: a custom signal had to
pass a millisecond value to new SnapshotStore(...) that happened to
match the windows of whichever ConcurrencySystem would drive it. Guess
too low and the signal silently contributed a narrower view of its
resource than every other signal, with nothing to warn about it - and
the value it needed was not reachable from anywhere in the interface.

LoadSignal.start() now receives a context with the widest window the
signal will be queried with (the wider of the gating and autoscaling
windows), so retention is derived rather than guessed. Signals built on
SnapshotStore apply it automatically via useSampleWindow(); hand-written
ones get the number explicitly. Implementations that declare no
parameter still satisfy the interface, so this breaks no custom signal.

Consequently retention stops being separately configurable: the
per-signal snapshotHistoryMillis options are gone, the Snapshotter no
longer threads a retention value into four constructors, and
snapshotHistorySecs now means one thing - the autoscaling window, which
signals size themselves to. This also removes the possibility of a
gating window wider than retention silently collapsing the two windows
into one, since retention covers the wider of them by construction.

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

@@ -799,10 +799,15 @@ export interface LoadSignal {
     getSample(sampleDurationMillis?: number): LoadSnapshot[];
     readonly name: string;
     readonly overloadedRatio: number;
-    start(): Promise<void>;
+    start(context: LoadSignalStartContext): Promise<void>;
     stop(): Promise<void>;
 }
 
+// @public
+export interface LoadSignalStartContext {
+    maxSampleWindowMillis: number;
+}
+
 // @public
 export interface LoadSnapshot {
     // (undocumented)
@@ -1546,7 +1551,6 @@ export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
         overloadedRatio: number;
         events: EventManager;
         event: EventTypeName;
-        snapshotHistoryMillis?: number;
         handler: (store: SnapshotStore<T>, payload: E) => void;
     }): Omit<LoadSignal, 'getSample'> & {
         store: SnapshotStore<T>;
@@ -1557,7 +1561,6 @@ export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
         name: string;
         overloadedRatio: number;
         intervalMillis: number;
-        snapshotHistoryMillis?: number;
         handler: (store: SnapshotStore<T>, intervalCallback: () => unknown) => void;
     }): Omit<LoadSignal, 'getSample'> & {
         store: SnapshotStore<T>;
@@ -1567,6 +1570,7 @@ export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
     getAll(): T[];
     getSample(sampleDurationMillis?: number): T[];
     push(snapshot: T, now?: Date): void;
+    useSampleWindow(maxSampleWindowMillis: number): void;
 }
 
 // @public (undocumented)

docs/upgrading/upgrading_v4.md

@@ -1349,6 +1349,31 @@ Overload is evaluated over two windows: a short one (`currentHistorySecs`, defau
 
 Both windows are now requested explicitly from every signal, so `snapshotHistorySecs` bounds the autoscaling window uniformly, custom signals included. With default settings nothing changes (the built-in signals retain exactly the 30s they are evaluated over, and `SnapshotStore` defaults to the same). You are only affected if a custom signal retains **more** history than `snapshotHistorySecs` — its stale snapshots no longer influence scaling — or if your custom `getSample()` implementation ignores its `sampleDurationMillis` argument, in which case it still contributes everything it has. Honour that argument (as `SnapshotStore` does) if you want the window respected.
 
+### `LoadSignal.start()` receives the sample window
+
+Retention used to be something a custom signal had to guess: you passed a millisecond value to `new SnapshotStore(...)` (or rolled your own storage) and hoped it matched the windows of the `ConcurrencySystem` that would end up driving it. Keeping less than the system samples meant silently contributing a narrower view of your resource than every other signal, with nothing to warn you.
+
+`start()` now receives a context carrying the widest window the signal will be queried with, so retention can be derived instead of guessed:
+
+```typescript
+import { SnapshotStore, type LoadSignal } from 'crawlee';
+
+const store = new SnapshotStore();
+
+const proxyHealthSignal: LoadSignal = {
+    name: 'proxyHealth',
+    overloadedRatio: 0.3,
+    async start({ maxSampleWindowMillis }) {
+        store.useSampleWindow(maxSampleWindowMillis);
+        // ...start collecting, calling store.push() per measurement
+    },
+    async stop() { /* ... */ },
+    getSample: (sampleDurationMillis) => store.getSample(sampleDurationMillis),
+};
+```
+
+Signals built through `SnapshotStore.fromInterval()` / `SnapshotStore.fromEvent()` do this for you. Existing implementations keep compiling — a `start()` that declares no parameters still satisfies the interface — they just retain whatever they chose to. Correspondingly, the `snapshotHistoryMillis` option was removed from the internal per-signal option types, and `SnapshotterOptions.snapshotHistorySecs` now means "the autoscaling window" (which signals size their retention to) rather than a separately configured retention.
+
 ## Stagehand type narrowings
 
 A few Stagehand-specific option types were tightened:

packages/core/src/autoscaling/client_load_signal.ts

@@ -22,7 +22,6 @@ export interface ClientLoadSignalOptions {
     clientSnapshotIntervalSecs?: number;
     maxClientErrors?: number;
     overloadedRatio?: number;
-    snapshotHistoryMillis?: number;
 }
 
 /**
@@ -37,7 +36,6 @@ export function createClientLoadSignal(options: ClientLoadSignalOptions) {
         name: 'clientInfo',
         overloadedRatio: options.overloadedRatio ?? 0.3,
         intervalMillis: (options.clientSnapshotIntervalSecs ?? 1) * 1000,
-        snapshotHistoryMillis: options.snapshotHistoryMillis,
         handler(store, intervalCallback) {
             const now = new Date();
 

packages/core/src/autoscaling/concurrency_system.ts

@@ -6,10 +6,10 @@ import { betterClearInterval, betterSetInterval } from '@apify/utilities';
 import type { CrawleeLogger } from '../log.js';
 import { serviceLocator } from '../service_locator.js';
 import type { LoadSignal } from './load_signal.js';
-import { Snapshotter } from './snapshotter.js';
+import { DEFAULT_SNAPSHOT_HISTORY_SECS, Snapshotter } from './snapshotter.js';
 import type { SnapshotterOptions } from './snapshotter.js';
 import type { SystemInfo } from './system_status.js';
-import { SystemStatus } from './system_status.js';
+import { DEFAULT_CURRENT_HISTORY_SECS, SystemStatus } from './system_status.js';
 
 export interface ConcurrencySystemOptions {
     /**
@@ -184,6 +184,12 @@ export class ConcurrencySystem implements IConcurrencySystem {
 
     private readonly snapshotter: Snapshotter;
     private readonly loadSignals: LoadSignal[];
+
+    /**
+     * The longest window any signal will be sampled over, handed to each of them at
+     * {@apilink ConcurrencySystem.start|`start()`} so they retain exactly that much history.
+     */
+    private readonly maxSampleWindowMillis: number;
     private readonly systemStatus: SystemStatus;
 
     private autoscaleInterval?: BetterIntervalID;
@@ -253,10 +259,18 @@ export class ConcurrencySystem implements IConcurrencySystem {
             snapshotter: this.snapshotter,
             loadSignals,
             currentHistorySecs,
-            // The historical (autoscaling) window matches what the built-in signals retain, and is requested
-            // explicitly so custom signals with a longer memory don't widen it.
+            // Both windows are requested from the signals explicitly, so a signal's own retention can neither widen
+            // nor (given the start context below) narrow what it contributes.
             historySecs: snapshotterOptions?.snapshotHistorySecs,
         });
+
+        // Signals are told how much history to keep when they start: exactly the longest window they will be sampled
+        // over, so nobody has to guess a retention value that matches this system's configuration.
+        this.maxSampleWindowMillis =
+            Math.max(
+                currentHistorySecs ?? DEFAULT_CURRENT_HISTORY_SECS,
+                snapshotterOptions?.snapshotHistorySecs ?? DEFAULT_SNAPSHOT_HISTORY_SECS,
+            ) * 1000;
     }
 
     /**
@@ -328,8 +342,9 @@ export class ConcurrencySystem implements IConcurrencySystem {
         }
         this.running = true;
 
-        await this.snapshotter.start();
-        await Promise.all(this.loadSignals.map(async (s) => s.start()));
+        const startContext = { maxSampleWindowMillis: this.maxSampleWindowMillis };
+        await this.snapshotter.start(startContext);
+        await Promise.all(this.loadSignals.map(async (s) => s.start(startContext)));
 
         this.autoscaleInterval = betterSetInterval(this._autoscale, this.autoscaleIntervalMillis);
 

packages/core/src/autoscaling/cpu_load_signal.ts

@@ -20,7 +20,6 @@ export interface CpuSnapshot extends LoadSnapshot {
  */
 export interface CpuLoadSignalOptions {
     overloadedRatio?: number;
-    snapshotHistoryMillis?: number;
     config: Configuration;
 }
 
@@ -35,7 +34,6 @@ export function createCpuLoadSignal(options: CpuLoadSignalOptions) {
         overloadedRatio: options.overloadedRatio ?? 0.4,
         events: serviceLocator.getEventManager(),
         event: EventType.SYSTEM_INFO,
-        snapshotHistoryMillis: options.snapshotHistoryMillis,
         handler(store, systemInfo) {
             const { cpuCurrentUsage, isCpuOverloaded } = systemInfo;
             const createdAt = systemInfo.createdAt ? new Date(systemInfo.createdAt) : new Date();

packages/core/src/autoscaling/event_loop_load_signal.ts

@@ -17,7 +17,6 @@ export interface EventLoopLoadSignalOptions {
     eventLoopSnapshotIntervalSecs?: number;
     maxBlockedMillis?: number;
     overloadedRatio?: number;
-    snapshotHistoryMillis?: number;
 }
 
 /**
@@ -33,7 +32,6 @@ export function createEventLoopLoadSignal(options: EventLoopLoadSignalOptions =
         name: 'eventLoopInfo',
         overloadedRatio: options.overloadedRatio ?? 0.6,
         intervalMillis,
-        snapshotHistoryMillis: options.snapshotHistoryMillis,
         handler(store, intervalCallback) {
             const now = new Date();
 

packages/core/src/autoscaling/load_signal.ts

@@ -14,6 +14,21 @@ export interface LoadSnapshot {
     isOverloaded: boolean;
 }
 
+/**
+ * Handed to a {@apilink LoadSignal} when it starts, so it can size its snapshot retention to what it will actually
+ * be asked for — without having to know how the {@apilink ConcurrencySystem} that drives it is configured.
+ */
+export interface LoadSignalStartContext {
+    /**
+     * The longest sample window the signal will be queried with (the wider of the task-gating and autoscaling
+     * windows). Keeping less history than this means contributing a narrower view of the resource than the other
+     * signals; keeping more is wasted memory, as the extra snapshots are never sampled.
+     *
+     * Signals built on {@apilink SnapshotStore} get this applied automatically.
+     */
+    maxSampleWindowMillis: number;
+}
+
 /**
  * A signal that reports whether a particular resource is overloaded.
  *
@@ -34,10 +49,14 @@ export interface LoadSignal {
      */
     readonly overloadedRatio: number;
 
-    /** Start collecting snapshots. Called when the pool starts. */
-    start(): Promise<void>;
+    /**
+     * Start collecting snapshots. Called when the {@apilink ConcurrencySystem} starts, with the sample window the
+     * signal should retain at least as much history as. Implementations that manage their own storage may ignore the
+     * argument, at the cost of contributing a window that doesn't match the other signals.
+     */
+    start(context: LoadSignalStartContext): Promise<void>;
 
-    /** Stop collecting snapshots. Called when the pool shuts down. */
+    /** Stop collecting snapshots. Called when the {@apilink ConcurrencySystem} shuts down. */
     stop(): Promise<void>;
 
     /**
@@ -53,12 +72,22 @@ export interface LoadSignal {
  */
 export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
     private snapshots: T[] = [];
-    private readonly historyMillis: number;
+    private historyMillis: number;
 
     constructor(historyMillis = 30_000) {
         this.historyMillis = historyMillis;
     }
 
+    /**
+     * Sizes retention to the window the signal will be sampled over, as handed to it in
+     * {@apilink LoadSignal.start|`start()`} — keep exactly what will be asked for, no more. Signals created through
+     * {@apilink SnapshotStore.fromInterval|`fromInterval()`}/{@apilink SnapshotStore.fromEvent|`fromEvent()`} call
+     * this for you; call it yourself when composing a store into a hand-written signal.
+     */
+    useSampleWindow(maxSampleWindowMillis: number): void {
+        this.historyMillis = maxSampleWindowMillis;
+    }
+
     /**
      * Add a snapshot and prune entries older than the history window.
      */
@@ -116,14 +145,13 @@ export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
         name: string;
         overloadedRatio: number;
         intervalMillis: number;
-        snapshotHistoryMillis?: number;
         handler: (store: SnapshotStore<T>, intervalCallback: () => unknown) => void;
     }): Omit<LoadSignal, 'getSample'> & {
         store: SnapshotStore<T>;
         handle: (cb: () => unknown) => void;
         getSample(sampleDurationMillis?: number): T[];
     } {
-        const store = new SnapshotStore<T>(options.snapshotHistoryMillis);
+        const store = new SnapshotStore<T>();
         let interval: BetterIntervalID = null!;
 
         const handle = (cb: () => unknown) => options.handler(store, cb);
@@ -134,7 +162,8 @@ export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
             store,
             handle,
             getSample: (ms) => store.getSample(ms),
-            async start() {
+            async start({ maxSampleWindowMillis }) {
+                store.useSampleWindow(maxSampleWindowMillis);
                 interval = betterSetInterval(handle, options.intervalMillis);
             },
             async stop() {
@@ -154,14 +183,13 @@ export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
         overloadedRatio: number;
         events: EventManager;
         event: EventTypeName;
-        snapshotHistoryMillis?: number;
         handler: (store: SnapshotStore<T>, payload: E) => void;
     }): Omit<LoadSignal, 'getSample'> & {
         store: SnapshotStore<T>;
         handle: (payload: E) => void;
         getSample(sampleDurationMillis?: number): T[];
     } {
-        const store = new SnapshotStore<T>(options.snapshotHistoryMillis);
+        const store = new SnapshotStore<T>();
 
         const handle = (payload: E) => options.handler(store, payload);
 
@@ -171,7 +199,8 @@ export class SnapshotStore<T extends LoadSnapshot = LoadSnapshot> {
             store,
             handle,
             getSample: (ms) => store.getSample(ms),
-            async start() {
+            async start({ maxSampleWindowMillis }) {
+                store.useSampleWindow(maxSampleWindowMillis);
                 options.events.on(options.event, handle);
             },
             async stop() {

packages/core/src/autoscaling/memory_load_signal.ts

@@ -5,7 +5,7 @@ import type { CrawleeLogger } from '../log.js';
 import { serviceLocator } from '../service_locator.js';
 import { getMemoryInfo } from '../system-info/memory-info.js';
 import { isContainerized } from '../system-info/runtime.js';
-import type { LoadSignal, LoadSnapshot } from './load_signal.js';
+import type { LoadSignal, LoadSignalStartContext, LoadSnapshot } from './load_signal.js';
 import { SnapshotStore } from './load_signal.js';
 import type { SystemInfo } from './system_status.js';
 
@@ -27,7 +27,6 @@ export interface MemorySnapshot extends LoadSnapshot {
 export interface MemoryLoadSignalOptions {
     maxUsedMemoryRatio?: number;
     overloadedRatio?: number;
-    snapshotHistoryMillis?: number;
     config: Configuration;
     log?: CrawleeLogger;
 }
@@ -51,7 +50,7 @@ export class MemoryLoadSignal implements LoadSignal {
     private lastLoggedCriticalMemoryOverloadAt: Date | null = null;
 
     constructor(options: MemoryLoadSignalOptions) {
-        this.store = new SnapshotStore(options.snapshotHistoryMillis);
+        this.store = new SnapshotStore();
         this.config = options.config;
         this.events = serviceLocator.getEventManager();
         this.log = options.log ?? serviceLocator.getLogger().child({ prefix: 'MemoryLoadSignal' });
@@ -60,7 +59,9 @@ export class MemoryLoadSignal implements LoadSignal {
         this._onSystemInfo = this._onSystemInfo.bind(this);
     }
 
-    async start(): Promise<void> {
+    async start({ maxSampleWindowMillis }: LoadSignalStartContext): Promise<void> {
+        this.store.useSampleWindow(maxSampleWindowMillis);
+
         const memoryMbytes = this.config.memoryMbytes ?? 0;
 
         if (memoryMbytes > 0) {