[pull] master from apify:master

packages/core/src/configuration.ts

@@ -13,6 +13,8 @@ import { type EventManager, LocalEventManager } from './events';
 import type { StorageManager } from './storages';
 import { type Constructor, entries } from './typedefs';
 
+const memoryStorages = new Map<string, MemoryStorage>();
+
 export interface ConfigurationOptions {
     /**
      * Defines storage client to be used.
@@ -438,20 +440,24 @@ export class Configuration {
      * @internal
      */
     createMemoryStorage(options: MemoryStorageOptions = {}): MemoryStorage {
-        const cacheKey = `MemoryStorage-${JSON.stringify(options)}`;
-
-        if (this.services.has(cacheKey)) {
-            return this.services.get(cacheKey) as MemoryStorage;
-        }
-
-        const storage = new MemoryStorage({
-            persistStorage: this.get('persistStorage'),
+        const storageOptions = {
+            persistStorage: this.get('persistStorage') as boolean,
             // Override persistStorage if user provides it via storageClientOptions
             ...options,
-        });
-        this.services.set(cacheKey, storage);
+        };
+
+        const cacheKey = `MemoryStorage-${JSON.stringify(storageOptions)}-${process.env.CRAWLEE_STORAGE_DIR ?? ''}`;
+
+        // Clients writing to the same directory are not independent, so they are shared between
+        // `Configuration` instances - otherwise they would purge each other's data mid-run.
+        // Purely in-memory clients have no such conflict and stay isolated per configuration.
+        const cache = storageOptions.persistStorage ? memoryStorages : this.services;
+
+        if (!cache.has(cacheKey)) {
+            cache.set(cacheKey, new MemoryStorage(storageOptions));
+        }
 
-        return storage;
+        return cache.get(cacheKey) as MemoryStorage;
     }
 
     useStorageClient(client: StorageClient): void {
@@ -498,6 +504,7 @@ export class Configuration {
      */
     static resetGlobalState(): void {
         delete this.globalConfig;
+        memoryStorages.clear();
     }
 
     protected buildOptions(options: ConfigurationOptions) {

packages/core/src/storages/utils.ts

@@ -57,13 +57,23 @@ export async function purgeDefaultStorages(
     const { config = Configuration.getGlobalConfig(), onlyPurgeOnce = false } = options;
     ({ client = config.getStorageClient() } = options);
 
-    const casted = client as StorageClient & { __purged?: boolean };
+    const casted = client as StorageClient & { __purged?: Promise<void> };
+
+    const runPurge = async () => {
+        try {
+            await casted.purge?.();
+        } catch (e) {
+            casted.__purged = undefined;
+            throw e;
+        }
+    };
 
     // if `onlyPurgeOnce` is true, will purge anytime this function is called, otherwise - only on start
     if (!onlyPurgeOnce || (config.get('purgeOnStart') && !casted.__purged)) {
-        casted.__purged = true;
-        await casted.purge?.();
+        casted.__purged = runPurge();
     }
+
+    await casted.__purged;
 }
 
 export interface UseStateOptions {

test/core/storages/utils.test.ts

@@ -1,5 +1,6 @@
 import type { Dictionary } from '@crawlee/core';
-import { Configuration, KeyValueStore, useState } from '@crawlee/core';
+import { Configuration, KeyValueStore, purgeDefaultStorages, useState } from '@crawlee/core';
+import type { StorageClient } from '@crawlee/types';
 
 import { MemoryStorageEmulator } from '../../shared/MemoryStorageEmulator';
 
@@ -64,3 +65,54 @@ describe('useState', () => {
         await manager.close();
     });
 });
+
+describe('purgeDefaultStorages', () => {
+    it('makes concurrent callers wait for an in-flight purge', async () => {
+        let purging = false;
+        let purgedWhileAlreadyPurging = false;
+
+        const client = {
+            async purge() {
+                purging = true;
+                await new Promise((resolve) => setTimeout(resolve, 50));
+                purging = false;
+            },
+        } as unknown as StorageClient;
+
+        const config = new Configuration({ purgeOnStart: true });
+
+        await Promise.all(
+            Array.from({ length: 3 }, async () => {
+                await purgeDefaultStorages({ onlyPurgeOnce: true, client, config });
+                if (purging) {
+                    purgedWhileAlreadyPurging = true;
+                }
+            }),
+        );
+
+        expect(purgedWhileAlreadyPurging).toBe(false);
+    });
+
+    it('purges only once for configurations persisting to the same directory', async () => {
+        const configA = new Configuration({ purgeOnStart: true, persistStorage: true });
+        const configB = new Configuration({ purgeOnStart: true, persistStorage: true });
+
+        const client = configA.getStorageClient();
+        expect(configB.getStorageClient()).toBe(client);
+
+        let purgeCount = 0;
+        client.purge = async () => void purgeCount++;
+
+        await purgeDefaultStorages({ onlyPurgeOnce: true, client, config: configA });
+        await purgeDefaultStorages({ onlyPurgeOnce: true, client: configB.getStorageClient(), config: configB });
+
+        expect(purgeCount).toBe(1);
+    });
+
+    it('keeps in-memory storage clients isolated per configuration', () => {
+        const configA = new Configuration({ persistStorage: false });
+        const configB = new Configuration({ persistStorage: false });
+
+        expect(configA.getStorageClient()).not.toBe(configB.getStorageClient());
+    });
+});