fix: wait for persist state listeners to run when event manager closes (#1481)

Author: vladfrangu

packages/core/src/events/event_manager.ts

@@ -58,6 +58,9 @@ export abstract class EventManager {
 
  // Emit final PERSIST_STATE event
  this.emit(EventType.PERSIST_STATE, { isMigrating: false });
+
+ // Wait for PERSIST_STATE to process
+ await this.waitForAllListenersToComplete();
  }
 
  on(event: EventTypeName, listener: (...args: any[]) => any): void {

test/e2e/automatic-persist-value/actor/.gitignore

@@ -0,0 +1,7 @@
+.idea
+.DS_Store
+node_modules
+package-lock.json
+apify_storage
+crawlee_storage
+storage

test/e2e/automatic-persist-value/actor/Dockerfile

@@ -0,0 +1,15 @@
+FROM apify/actor-node:16
+
+COPY packages ./packages
+COPY package*.json ./
+
+RUN npm --quiet set progress=false \
+ && npm install --only=prod --no-optional \
+ && echo "Installed NPM packages:" \
+ && (npm list --only=prod --no-optional --all || true) \
+ && echo "Node.js version:" \
+ && node --version \
+ && echo "NPM version:" \
+ && npm --version
+
+COPY . ./

test/e2e/automatic-persist-value/actor/apify.json

@@ -0,0 +1,6 @@
+{
+"name": "test-automatic-persist-value",
+"version": "0.0",
+"buildTag": "latest",
+"env": null
+}

test/e2e/automatic-persist-value/actor/main.js

@@ -0,0 +1,22 @@
+import { Actor, KeyValueStore } from 'apify';
+import { ApifyStorageLocal } from '@apify/storage-local';
+import { BasicCrawler } from '@crawlee/basic';
+
+const mainOptions = {
+ exit: Actor.isAtHome(),
+ storage: process.env.STORAGE_IMPLEMENTATION === 'LOCAL' ? new ApifyStorageLocal() : undefined,
+};
+
+await Actor.main(async () => {
+ const kv = await KeyValueStore.open('test');
+
+ const crawler = new BasicCrawler({
+ async requestHandler() {
+ const automaticValue = await kv.getAutoSavedValue('hello');
+
+ automaticValue.crawlee = 'awesome!';
+ },
+ });
+
+ await crawler.run(['https://example.com']);
+}, mainOptions);

test/e2e/automatic-persist-value/actor/package.json

@@ -0,0 +1,17 @@
+{
+"name": "test-automatic-persist-value",
+"version": "0.0.1",
+"description": "Key-Value Store - Automatic Persist Value Test",
+"dependencies": {
+ "apify": "^3.0.0-beta.75",
+ "@apify/storage-local": "^2.1.0",
+ "@crawlee/basic": "file:./packages/basic-crawler",
+ "@crawlee/core": "file:./packages/core",
+ "@crawlee/memory-storage": "file:./packages/memory-storage",
+ "@crawlee/types": "file:./packages/types",
+ "@crawlee/utils": "file:./packages/utils"
+ },
+ "scripts": { "start": "node main.js" },
+"type": "module",
+"license": "ISC"
+}

test/e2e/automatic-persist-value/test.mjs

@@ -0,0 +1,19 @@
+import { initialize, expect, getActorTestDir, runActor, getKeyValueStoreItems } from '../tools.mjs';
+
+const testActorDirname = getActorTestDir(import.meta.url);
+await initialize(testActorDirname);
+
+await runActor(testActorDirname);
+
+const kvsItems = await getKeyValueStoreItems(testActorDirname, 'test');
+
+expect(kvsItems.length === 1, 'Key-value store automatically saved the value expected to be auto-saved');
+
+const [{ name, raw }] = kvsItems;
+
+expect(name === 'crawlee', 'Key-value store auto-saved value is named "crawlee"');
+
+const parsed = JSON.parse(raw.toString());
+
+expect(typeof parsed === 'object' && parsed !== null, 'Key-value store auto-save value is a non-nullable object');
+expect(parsed.crawlee === 'awesome!', 'Key-value store auto-save value has a property "crawlee" that is set to "awesome!"');

test/e2e/autoscaling-max-tasks-per-minute/actor/package.json

@@ -6,7 +6,6 @@
  "apify": "^3.0.0-beta.75",
  "@apify/storage-local": "^2.1.0",
  "@crawlee/basic": "file:./packages/basic-crawler",
- "@crawlee/browser-pool": "file:./packages/browser-pool",
  "@crawlee/core": "file:./packages/core",
  "@crawlee/memory-storage": "file:./packages/memory-storage",
  "@crawlee/types": "file:./packages/types",

test/e2e/tools.mjs

@@ -1,12 +1,13 @@
 import { dirname, join, resolve } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { existsSync } from 'node:fs';
-import { readdir } from 'node:fs/promises';
+import { readdir, readFile } from 'node:fs/promises';
 import { homedir } from 'node:os';
 import { setTimeout } from 'node:timers/promises';
 import { execSync } from 'node:child_process';
 import fs from 'fs-extra';
 import { Actor } from 'apify';
+// eslint-disable-next-line import/no-relative-packages
 import { URL_NO_COMMAS_REGEX } from '../../packages/utils/dist/index.mjs';
 
 export const SKIPPED_TEST_CLOSE_CODE = 404;
@@ -207,6 +208,35 @@ export async function getDatasetItems(dirName) {
  return datasetItems;
 }
 
+/**
+ * Gets all items in the key-value store, as a Buffer
+ * @param {string} dirName
+ * @param {string} storeName
+ */
+export async function getKeyValueStoreItems(dirName, storeName) {
+ const dir = getStorage(dirName);
+ const storePath = join(dir, `key_value_stores/${storeName}`);
+
+ if (!existsSync(storePath)) {
+ return [];
+ }
+
+ const dirents = await readdir(storePath, { withFileTypes: true });
+ const fileNames = dirents.filter((dirent) => dirent.isFile());
+ const keyValueStoreRecords = [];
+
+ for (const fileName of fileNames) {
+ if (fileName.name.includes('__metadata__')) continue;
+
+ const filePath = join(storePath, fileName.name);
+ const buffer = await readFile(filePath);
+
+ keyValueStoreRecords.push({ name: fileName.name.split('.').slice(0, -1).join('.'), raw: buffer });
+ }
+
+ return keyValueStoreRecords;
+}
+
 /**
  * @param {string} dirName
  */