fix: key value stores emitting an error when multiple write promises ran in parallel (#1460)

Author: vladfrangu

packages/memory-storage/src/memory-storage.ts

@@ -2,7 +2,8 @@
 import type * as storage from '@crawlee/types';
 import type { Dictionary } from '@crawlee/types';
 import { s } from '@sapphire/shapeshift';
-import { ensureDir, pathExistsSync } from 'fs-extra';
+import { ensureDirSync, pathExistsSync } from 'fs-extra';
+import { renameSync } from 'node:fs';
 import { rm, rename, readdir } from 'node:fs/promises';
 import { resolve } from 'node:path';
 import { DatasetClient } from './resource-clients/dataset';
@@ -114,6 +115,20 @@ export class MemoryStorage implements storage.StorageClient {
  * - local directory containing the default request queue.
  */
  async purge(): Promise<void> {
+ // Key-value stores
+ const keyValueStores = await readdir(this.keyValueStoresDirectory).catch(() => []);
+ const keyValueStorePromises: Promise<void>[] = [];
+
+ for (const keyValueStoreFolder of keyValueStores) {
+ if (keyValueStoreFolder.startsWith('__CRAWLEE_TEMPORARY') || keyValueStoreFolder.startsWith('__OLD')) {
+ keyValueStorePromises.push((await this.batchRemoveFiles(resolve(this.keyValueStoresDirectory, keyValueStoreFolder)))());
+ } else if (keyValueStoreFolder === 'default') {
+ keyValueStorePromises.push(this.handleDefaultKeyValueStore(resolve(this.keyValueStoresDirectory, keyValueStoreFolder))());
+ }
+ }
+
+ void Promise.allSettled(keyValueStorePromises);
+
  // Datasets
  const datasets = await readdir(this.datasetsDirectory).catch(() => []);
  const datasetPromises: Promise<void>[] = [];
@@ -137,20 +152,6 @@ export class MemoryStorage implements storage.StorageClient {
  }
 
  void Promise.allSettled(requestQueuePromises);
-
- // Key-value stores
- const keyValueStores = await readdir(this.keyValueStoresDirectory).catch(() => []);
- const keyValueStorePromises: Promise<void>[] = [];
-
- for (const keyValueStoreFolder of keyValueStores) {
- if (keyValueStoreFolder.startsWith('__CRAWLEE_TEMPORARY')) {
- keyValueStorePromises.push((await this.batchRemoveFiles(resolve(this.keyValueStoresDirectory, keyValueStoreFolder)))());
- } else if (keyValueStoreFolder === 'default') {
- keyValueStorePromises.push((await this.handleDefaultKeyValueStore(resolve(this.keyValueStoresDirectory, keyValueStoreFolder)))());
- }
- }
-
- void Promise.allSettled(keyValueStorePromises);
  }
 
  /**
@@ -162,7 +163,7 @@ export class MemoryStorage implements storage.StorageClient {
  await Promise.all(promises);
  }
 
- private async handleDefaultKeyValueStore(folder: string): Promise<() => Promise<void>> {
+ private handleDefaultKeyValueStore(folder: string): () => Promise<void> {
  const storagePathExists = pathExistsSync(folder);
  const temporaryPath = resolve(folder, '../__CRAWLEE_MIGRATING_KEY_VALUE_STORE__');
 
@@ -176,26 +177,36 @@ export class MemoryStorage implements storage.StorageClient {
 
  if (storagePathExists) {
  // Create temporary folder to save important files in
- await ensureDir(temporaryPath);
+ ensureDirSync(temporaryPath);
 
  // Go through each file and save the ones that are important
  for (const entity of possibleInputKeys) {
  const originalFilePath = resolve(folder, entity);
  const tempFilePath = resolve(temporaryPath, entity);
 
  try {
- await rename(originalFilePath, tempFilePath);
+ renameSync(originalFilePath, tempFilePath);
  } catch {
  // Ignore
  }
  }
 
  // Remove the original folder and all its content
- const tempPathForOldFolder = resolve(folder, '../__OLD_DEFAULT__');
- await rename(folder, tempPathForOldFolder);
+ let counter = 0;
+ let tempPathForOldFolder = resolve(folder, `../__OLD_DEFAULT_${counter}__`);
+ let done = false;
+
+ while (!done) {
+ try {
+ renameSync(folder, tempPathForOldFolder);
+ done = true;
+ } catch {
+ tempPathForOldFolder = resolve(folder, `../__OLD_DEFAULT_${++counter}__`);
+ }
+ }
 
  // Replace the temporary folder with the original folder
- await rename(temporaryPath, folder);
+ renameSync(temporaryPath, folder);
 
  // Remove the old folder
  return async () => (await this.batchRemoveFiles(tempPathForOldFolder))();