From 8e8c75beaa6dcb5c7fe671fbc8fad430fd7af516 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jind=C5=99ich=20B=C3=A4r?= Date: Fri, 7 Aug 2026 13:48:17 +0200 Subject: [PATCH] fix(core): do not purge storages that are already in use (#3988) `purgeDefaultStorages` marked the client as purged before the purge actually finished, and every `Configuration` built its own storage client for the same directory, so a purge could wipe the request queue of a run already in progress. Closes #3156 --- packages/core/src/configuration.ts | 29 ++++++++++------ packages/core/src/storages/utils.ts | 16 +++++++-- test/core/storages/utils.test.ts | 54 ++++++++++++++++++++++++++++- 3 files changed, 84 insertions(+), 15 deletions(-) diff --git a/packages/core/src/configuration.ts b/packages/core/src/configuration.ts index 42b9c22db4b0..60ec71551676 100644 --- a/packages/core/src/configuration.ts +++ b/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(); + 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) { diff --git a/packages/core/src/storages/utils.ts b/packages/core/src/storages/utils.ts index 31135c948dd7..0422b769da40 100644 --- a/packages/core/src/storages/utils.ts +++ b/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 }; + + 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 { diff --git a/test/core/storages/utils.test.ts b/test/core/storages/utils.test.ts index 3cde7aeed015..29ff7e949da9 100644 --- a/test/core/storages/utils.test.ts +++ b/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()); + }); +});