fix(Storage): queue up opening storages to prevent issues in concurre…

…nt calls (#1865)

packages/core/package.json

@@ -62,6 +62,7 @@
         "@crawlee/memory-storage": "^3.3.0",
         "@crawlee/types": "^3.3.0",
         "@crawlee/utils": "^3.3.0",
+        "@sapphire/async-queue": "^1.5.0",
         "@types/tough-cookie": "^4.0.2",
         "@vladfrangu/async_event_emitter": "^2.0.0",
         "csv-stringify": "^6.2.0",

packages/core/src/storages/storage_manager.ts

@@ -1,4 +1,5 @@
 import type { Dictionary, StorageClient } from '@crawlee/types';
+import { AsyncQueue } from '@sapphire/async-queue';
 import { Configuration } from '../configuration';
 import type { Constructor } from '../typedefs';
 
@@ -22,6 +23,7 @@ export class StorageManager<T extends IStorage = IStorage> {
     private readonly name: 'Dataset' | 'KeyValueStore' | 'RequestQueue';
     private readonly StorageConstructor: Constructor<T> & { name: string };
     private readonly cache = new Map<string, T>();
+    private readonly storageOpenQueue = new AsyncQueue();
 
     constructor(
         StorageConstructor: Constructor<T>,
@@ -65,6 +67,8 @@ export class StorageManager<T extends IStorage = IStorage> {
     }
 
     async openStorage(idOrName?: string | null, client?: StorageClient): Promise<T> {
+        await this.storageOpenQueue.wait();
+
         if (!idOrName) {
             const defaultIdConfigKey = DEFAULT_ID_CONFIG_KEYS[this.name];
             idOrName = this.config.get(defaultIdConfigKey) as string;
@@ -84,6 +88,8 @@ export class StorageManager<T extends IStorage = IStorage> {
             this._addStorageToCache(storage);
         }
 
+        this.storageOpenQueue.shift();
+
         return storage;
     }
 

yarn.lock

@@ -832,6 +832,7 @@ __metadata:
     "@crawlee/memory-storage": ^3.3.0
     "@crawlee/types": ^3.3.0
     "@crawlee/utils": ^3.3.0
+    "@sapphire/async-queue": ^1.5.0
     "@types/tough-cookie": ^4.0.2
     "@vladfrangu/async_event_emitter": ^2.0.0
     csv-stringify: ^6.2.0