feat: Dynamic memory snapshots (#3471)

### Description

- Allow `Snapshotter` to take into account changes in total memory. 
- When fixed memory through `Configuration.memoryMbytes` is not set,
then dynamic memory mode based on `Configuration.availableMemoryRatio`
is used. If the event manager is sending sufficient information (total
memory available in the system), then the `Snapshotter` is capable of
scaling with the total available memory. (Otherwise, fall back to
previous behavior - fixed memory based on initial measurement.)

(Difference to Python implementation -> No dedicated Ratio type. Since
in JS implementation, there is no user-facing change, there is also no
need for this new type. Instead, the ratio is saved as private property
of Snapshotter when needed.)

### Issues

- Closes: #3408 

### Testing

- Unit test

### Checklist

- [ ] CI passed

---------

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Jindřich Bär <jindrichbar@gmail.com>

Author: 3 people

packages/core/src/autoscaling/snapshotter.ts

@@ -127,6 +127,7 @@ export class Snapshotter {
     maxUsedMemoryRatio: number;
     maxClientErrors: number;
     maxMemoryBytes!: number;
+    private maxMemoryRatio: number | undefined;
 
     cpuSnapshots: CpuSnapshot[] = [];
     eventLoopSnapshots: EventLoopSnapshot[] = [];
@@ -180,7 +181,6 @@ export class Snapshotter {
         this.maxBlockedMillis = maxBlockedMillis;
         this.maxUsedMemoryRatio = maxUsedMemoryRatio;
         this.maxClientErrors = maxClientErrors;
-
         // We need to pre-bind those functions to be able to successfully remove listeners.
         this._snapshotCpu = this._snapshotCpu.bind(this);
         this._snapshotMemory = this._snapshotMemory.bind(this);
@@ -195,22 +195,20 @@ export class Snapshotter {
         if (memoryMbytes > 0) {
             this.maxMemoryBytes = memoryMbytes * 1024 * 1024;
         } else {
-            let totalBytes: number;
-
-            if (this.config.get('systemInfoV2')) {
-                const containerized = this.config.get('containerized', await isContainerized());
-                const memInfo = await getMemoryInfoV2(containerized);
-                totalBytes = memInfo.totalBytes;
+            this.maxMemoryRatio = this.config.get('availableMemoryRatio');
+            if (!this.maxMemoryRatio) {
+                throw new Error('availableMemoryRatio is not set in configuration.');
             } else {
-                const memInfo = await getMemoryInfo();
-                totalBytes = memInfo.totalBytes;
+                this.log.debug(
+                    `Setting max memory of this run to ${this.maxMemoryRatio * 100} % of available memory. ` +
+                        'Use the CRAWLEE_MEMORY_MBYTES or CRAWLEE_AVAILABLE_MEMORY_RATIO environment variable to override it.',
+                );
             }
-
-            this.maxMemoryBytes = Math.ceil(totalBytes * this.config.get('availableMemoryRatio')!);
-            this.log.debug(
-                `Setting max memory of this run to ${Math.round(this.maxMemoryBytes / 1024 / 1024)} MB. ` +
-                    'Use the CRAWLEE_MEMORY_MBYTES or CRAWLEE_AVAILABLE_MEMORY_RATIO environment variable to override it.',
-            );
+            // Create a fallback memory measurement in case of missing memTotalBytes in SystemInfo. Weak types of
+            // SystemInfo do not guarantee that memTotalBytes is always present, and without it, we cannot compute the
+            // maxMemoryBytes.
+            // This does not happen in practice, but code allows it.
+            this.maxMemoryBytes = await this.getTotalMemoryBytes();
         }
 
         // Start snapshotting.
@@ -299,21 +297,27 @@ export class Snapshotter {
     protected _snapshotMemory(systemInfo: SystemInfo) {
         const createdAt = systemInfo.createdAt ? new Date(systemInfo.createdAt) : new Date();
         this._pruneSnapshots(this.memorySnapshots, createdAt);
-        const { memCurrentBytes } = systemInfo;
+        const { memCurrentBytes, memTotalBytes } = systemInfo;
+
+        let maxMemoryBytes = this.maxMemoryBytes!;
+        if (this.maxMemoryRatio !== undefined && this.maxMemoryRatio > 0) {
+            maxMemoryBytes = this.maxMemoryRatio * (memTotalBytes ?? this.maxMemoryBytes);
+        }
+
         const snapshot: MemorySnapshot = {
             createdAt,
-            isOverloaded: memCurrentBytes! / this.maxMemoryBytes! > this.maxUsedMemoryRatio,
+            isOverloaded: memCurrentBytes! / maxMemoryBytes > this.maxUsedMemoryRatio,
             usedBytes: memCurrentBytes,
         };
 
         this.memorySnapshots.push(snapshot);
-        this._memoryOverloadWarning(systemInfo);
+        this._memoryOverloadWarning(systemInfo, maxMemoryBytes);
     }
 
     /**
      * Checks for critical memory overload and logs it to the console.
      */
-    protected _memoryOverloadWarning(systemInfo: SystemInfo) {
+    protected _memoryOverloadWarning(systemInfo: SystemInfo, maxMemoryBytes: number) {
         const { memCurrentBytes } = systemInfo;
         const createdAt = systemInfo.createdAt ? new Date(systemInfo.createdAt) : new Date();
         if (
@@ -322,18 +326,18 @@ export class Snapshotter {
         )
             return;
 
-        const maxDesiredMemoryBytes = this.maxUsedMemoryRatio * this.maxMemoryBytes!;
-        const reserveMemory = this.maxMemoryBytes! * (1 - this.maxUsedMemoryRatio) * RESERVE_MEMORY_RATIO;
+        const maxDesiredMemoryBytes = this.maxUsedMemoryRatio * maxMemoryBytes;
+        const reserveMemory = maxMemoryBytes * (1 - this.maxUsedMemoryRatio) * RESERVE_MEMORY_RATIO;
         const criticalOverloadBytes = maxDesiredMemoryBytes + reserveMemory;
         const isCriticalOverload = memCurrentBytes! > criticalOverloadBytes;
 
         if (isCriticalOverload) {
-            const usedPercentage = Math.round((memCurrentBytes! / this.maxMemoryBytes!) * 100);
+            const usedPercentage = Math.round((memCurrentBytes! / maxMemoryBytes) * 100);
             const toMb = (bytes: number) => Math.round(bytes / 1024 ** 2);
             this.log.warning(
                 'Memory is critically overloaded. ' +
                     `Using ${toMb(memCurrentBytes!)} MB of ${toMb(
-                        this.maxMemoryBytes!,
+                        maxMemoryBytes,
                     )} MB (${usedPercentage}%). Consider increasing available memory.`,
             );
             this.lastLoggedCriticalMemoryOverloadAt = createdAt;
@@ -429,4 +433,12 @@ export class Snapshotter {
         }
         snapshots.splice(0, oldCount);
     }
+
+    protected async getTotalMemoryBytes() {
+        if (this.config.get('systemInfoV2')) {
+            const containerized = this.config.get('containerized', await isContainerized());
+            return (await getMemoryInfoV2(containerized)).totalBytes;
+        }
+        return (await getMemoryInfo()).totalBytes;
+    }
 }

packages/core/src/autoscaling/system_status.ts

@@ -14,6 +14,7 @@ export interface SystemInfo {
     eventLoopInfo: ClientInfo;
     cpuInfo: ClientInfo;
     clientInfo: ClientInfo;
+    memTotalBytes?: number;
     memCurrentBytes?: number;
     /**
      * Platform only property

packages/core/src/events/local_event_manager.ts

@@ -104,19 +104,21 @@ export class LocalEventManager extends EventManager {
 
     private async createMemoryInfo() {
         try {
-            if (this.config.get('systemInfoV2')) {
-                const memInfo = await getMemoryInfoV2(await this.isContainerizedWrapper());
-                return {
-                    memCurrentBytes: memInfo.mainProcessBytes + memInfo.childProcessesBytes,
-                };
-            }
-            const memInfo = await getMemoryInfo();
+            const memInfo = await this.getMemoryInfo();
             return {
+                memTotalBytes: memInfo.totalBytes,
                 memCurrentBytes: memInfo.mainProcessBytes + memInfo.childProcessesBytes,
             };
         } catch (err) {
             log.exception(err as Error, 'Memory snapshot failed.');
             return {};
         }
     }
+
+    private async getMemoryInfo() {
+        if (this.config.get('systemInfoV2')) {
+            return getMemoryInfoV2(await this.isContainerizedWrapper());
+        }
+        return getMemoryInfo();
+    }
 }

test/core/autoscaling/snapshotter.test.ts

@@ -2,12 +2,12 @@ import os from 'node:os';
 
 import { Configuration, EventType, LocalEventManager, Snapshotter } from '@crawlee/core';
 import type { MemoryInfo } from '@crawlee/utils';
-import * as utils from '@crawlee/utils';
 import { sleep } from '@crawlee/utils';
 
 import log from '@apify/log';
 
 const toBytes = (x: number) => x * 1024 * 1024;
+const noop = () => {};
 
 describe('Snapshotter', () => {
     let logLevel: number;
@@ -139,7 +139,6 @@ describe('Snapshotter', () => {
 
         cpusMock.mockReturnValue(fakeCpu as any);
 
-        const noop = () => {};
         const config = new Configuration({ maxUsedCpuRatio: 0.5 });
         const snapshotter = new Snapshotter({ config });
         // do not initialize the event intervals as we will fire them manually
@@ -178,7 +177,6 @@ describe('Snapshotter', () => {
     test('correctly marks eventLoopOverloaded', () => {
         const clock = vitest.useFakeTimers();
         try {
-            const noop = () => {};
             const snapshotter = new Snapshotter({ maxBlockedMillis: 5, eventLoopSnapshotIntervalSecs: 0 });
             // @ts-expect-error Calling protected method
             snapshotter._snapshotEventLoop(noop);
@@ -208,13 +206,12 @@ describe('Snapshotter', () => {
     });
 
     test('correctly marks memoryOverloaded using OS metrics', async () => {
-        const noop = () => {};
         const memoryData = {
             totalBytes: toBytes(10000),
             mainProcessBytes: toBytes(1000),
             childProcessesBytes: toBytes(1000),
         } as MemoryInfo;
-        vitest.spyOn(utils, 'getMemoryInfoV2').mockResolvedValue(memoryData);
+        vitest.spyOn(LocalEventManager.prototype as any, 'getMemoryInfo').mockResolvedValue(memoryData);
         const config = new Configuration({ availableMemoryRatio: 1 });
         const snapshotter = new Snapshotter({ config, maxUsedMemoryRatio: 0.5 });
         // do not initialize the event intervals as we will fire them manually
@@ -245,31 +242,48 @@ describe('Snapshotter', () => {
     });
 
     test('correctly logs critical memory overload', async () => {
-        vitest.spyOn(utils, 'getMemoryInfoV2').mockResolvedValueOnce({ totalBytes: toBytes(10000) } as MemoryInfo);
+        const initialMemory = toBytes(10000);
+        const usageRatio1 = 0.75; // below warning usage
+        const usageRatio2 = 0.76; // above warning usage
+        const memoryData: MemoryInfo = {
+            totalBytes: initialMemory,
+            freeBytes: initialMemory * (1 - usageRatio1),
+            usedBytes: initialMemory * usageRatio1,
+            mainProcessBytes: initialMemory * usageRatio1,
+            childProcessesBytes: 0,
+        };
+
+        // Mock memory info to be able to inject custom memory measurement data.
+        vitest.spyOn(LocalEventManager.prototype as any, 'getMemoryInfo').mockResolvedValue(memoryData);
         const config = new Configuration({ availableMemoryRatio: 1 });
         const snapshotter = new Snapshotter({ config, maxUsedMemoryRatio: 0.5 });
+
+        const eventManager = config.getEventManager() as LocalEventManager;
         await snapshotter.start();
         const warningSpy = vitest.spyOn(snapshotter.log, 'warning').mockImplementation(() => {});
 
-        // @ts-expect-error Calling private method
-        snapshotter._memoryOverloadWarning({
-            memCurrentBytes: toBytes(7600),
-        });
+        // First snapshot - below warning usage
+        await eventManager.emitSystemInfoEvent(noop);
+        expect(warningSpy).not.toBeCalled();
+
+        // Second snapshot - above warning usage
+        memoryData.usedBytes = initialMemory * usageRatio2;
+        memoryData.mainProcessBytes = initialMemory * usageRatio2;
+        memoryData.freeBytes = initialMemory * (1 - usageRatio2);
+        await eventManager.emitSystemInfoEvent(noop);
         expect(warningSpy).toBeCalled();
         warningSpy.mockReset();
 
-        // @ts-expect-error Calling private method
-        snapshotter._memoryOverloadWarning({
-            memCurrentBytes: toBytes(7500),
-        });
+        // Second snapshot again - repeated warning ignored
+        await eventManager.emitSystemInfoEvent(noop);
         expect(warningSpy).not.toBeCalled();
+        warningSpy.mockReset();
 
         vitest.restoreAllMocks();
         await snapshotter.stop();
     });
 
     test('correctly marks clientOverloaded', () => {
-        const noop = () => {};
         // mock client data
         const apifyClient = Configuration.getStorageClient();
         const oldStats = apifyClient.stats;
@@ -332,4 +346,61 @@ describe('Snapshotter', () => {
         expect(diffBetween).toBeLessThan(SAMPLE_SIZE_MILLIS);
         expect(diffWithin).toBeLessThan(SAMPLE_SIZE_MILLIS);
     });
+
+    test.each([
+        true,
+        false,
+    ])('correctly handles dynamic vs static memory limit when total memory changes (dynamic=%s)', async (dynamic) => {
+        /**
+         * Two memory snapshots are emitted with the same process memory usage but different total memory.
+         * First snapshot is overloaded in both modes. Using 60% of total memory, while the limit is 50% in both modes.
+         * Second snapshot doubles the total memory while keeping the same usage:
+         * - Dynamic mode (availableMemoryRatio): maxMemoryBytes should update → not overloaded
+         * - Static mode (memoryMbytes): maxMemoryBytes stays fixed → still overloaded
+         */
+        const initialTotalBytes = toBytes(100);
+        const allowedMemoryUsageRatio = 0.5;
+        const actualMemoryUsage = 0.6 * initialTotalBytes;
+
+        // Initial snapshot. Overloaded in both modes.
+        const memoryData: MemoryInfo = {
+            totalBytes: initialTotalBytes,
+            freeBytes: initialTotalBytes - actualMemoryUsage,
+            usedBytes: actualMemoryUsage,
+            mainProcessBytes: actualMemoryUsage,
+            childProcessesBytes: 0,
+        };
+
+        // Mock memory info to be able to inject custom memory measurement data.
+        vitest.spyOn(LocalEventManager.prototype as any, 'getMemoryInfo').mockResolvedValue(memoryData);
+
+        let config: Configuration;
+        if (dynamic) {
+            // Dynamic: Allow usage of 50 % of available memory through ratio
+            config = new Configuration({ availableMemoryRatio: allowedMemoryUsageRatio });
+        } else {
+            // Static: Allow usage of 50 % of available memory through fixed value
+            config = new Configuration({ memoryMbytes: (allowedMemoryUsageRatio * initialTotalBytes) / 1024 / 1024 });
+        }
+
+        const snapshotter = new Snapshotter({ config });
+        vitest.spyOn(LocalEventManager.prototype, 'init').mockImplementation(async () => {});
+        const eventManager = config.getEventManager() as LocalEventManager;
+        await snapshotter.start();
+
+        // First snapshot - full usage of the memory, should be overloaded in both modes
+        await eventManager.emitSystemInfoEvent(noop);
+
+        // Second snapshot - total memory doubled, should be overloaded only in static mode
+        memoryData.totalBytes = initialTotalBytes * 2;
+        memoryData.freeBytes = memoryData.totalBytes - actualMemoryUsage;
+        await eventManager.emitSystemInfoEvent(noop);
+
+        const memorySnapshots = snapshotter.getMemorySample();
+        expect(memorySnapshots).toHaveLength(2);
+        expect(memorySnapshots[0].isOverloaded).toBe(true);
+        expect(memorySnapshots[1].isOverloaded).toBe(!dynamic);
+
+        await snapshotter.stop();
+    });
 });