feat: Actor.charge() (#346)

* Add signatures of new methods * Add ACTOR_MAX_TOTAL_CHARGE_USD configuration option * Fix type error * Update method signatures to be more in line with Actor whitepaper * Update apify-client * Partially implement ChargingManager * Use ChargingManager in Actor * Read pricing info and use it * Dataset set up * Load more charging information on platform * Make sure that we stay within the budget when charging * Reorder stuff * Fill in docblocks * Reorder operations to prevent race conditions * Update apify-client * Update e2e test directory structure * WIP: Add e2e sdk test setup * Finalize sdk e2e testing environment * Initial e2e test of Actor.charge * Improve test runner * Make Actor.charge test fail * More tests * Simplify control flow * Ensure Actor init in PPE methods * Improve logging * Add remaining charges to ChargeResult * Do not charge for items not yet added to the dataset * Add getPricingInfo method * Allow switching actor into PPE mode locally for testing * Bug * Apply suggestions from code review Co-authored-by: Martin Adámek <banan23@gmail.com> * Extend helloWorld test * Update packages/apify/src/internals/charging.ts Co-authored-by: Martin Adámek <banan23@gmail.com> * Use an existing type for pricingModel * Make charging log dataset opt-in * Address code review comments * Address more review comments * Hide global Actor methods other than charge and add getChargingManager instead * Repeat ourselves to please our documentation tool * Remove unused import * Update e2e tests * Fix local charging * Lint * Do not allow using the charging log dataset on platform * Ignore a docblock * Log a single line for each Actor.charge * Make consts private --------- Co-authored-by: Martin Adámek <banan23@gmail.com>

Author: janbuchar

.gitignore

@@ -27,3 +27,5 @@ apify_storage
 crawlee_storage
 storage
 .turbo
+*.tgz
+mise.toml

package-lock.json

@@ -40,7 +40,10 @@
  "build": "turbo run build",
  "ci:build": "turbo run build --cache-dir=\".turbo\"",
  "test": "vitest run --silent",
- "test:e2e": "node test/e2e/run.mjs",
+ "test:e2e": "npm run test:e2e:scrapers && npm run test:e2e:sdk",
+ "test:e2e:scrapers": "node test/e2e/runScraperTests.mjs",
+ "test:e2e:sdk": "npm run test:e2e:sdk:tarball && node test/e2e/runSdkTests.mjs",
+ "test:e2e:sdk:tarball": "npm run build && cd packages/apify && mv $(npm pack | tail -n 1) ../../test/e2e/apify.tgz",
  "coverage": "vitest --coverage",
  "release": "npm run build && lerna publish from-package --contents dist",
  "publish:next": "lerna publish --canary --preid beta --dist-tag next",

package.json

@@ -62,7 +62,7 @@
  "@crawlee/core": "^3.9.0",
  "@crawlee/types": "^3.9.0",
  "@crawlee/utils": "^3.9.0",
- "apify-client": "^2.10.0",
+ "apify-client": "^2.11.1",
  "fs-extra": "^11.2.0",
  "ow": "^0.28.2",
  "semver": "^7.5.4",

packages/apify/package.json

@@ -37,6 +37,8 @@ import {
 import ow from 'ow';
 
 import { Configuration } from './configuration';
+import { ChargingManager } from './internals/charging';
+import type { ChargeOptions, ChargeResult } from './internals/charging';
 import { KeyValueStore } from './key_value_store';
 import { PlatformEventManager } from './platform_event_manager';
 import type { ProxyConfigurationOptions } from './proxy_configuration';
@@ -86,11 +88,14 @@ export class Actor<Data extends Dictionary = Dictionary> {
  */
  private isRebooting = false;
 
+ private chargingManager: ChargingManager;
+
  constructor(options: ConfigurationOptions = {}) {
  // use default configuration object if nothing overridden (it fallbacks to env vars)
  this.config = Object.keys(options).length === 0 ? Configuration.getGlobalConfig() : new Configuration(options);
  this.apifyClient = this.newClient();
  this.eventManager = new PlatformEventManager(this.config);
+ this.chargingManager = new ChargingManager(this.config, this.apifyClient);
  }
 
  /**
@@ -222,6 +227,9 @@ export class Actor<Data extends Dictionary = Dictionary> {
  log.debug(`Default storages purged`);
 
  Configuration.storage.enterWith(this.config);
+
+ await this.chargingManager.init();
+ log.debug(`ChargingManager initialized`, this.chargingManager.getPricingInfo());
  }
 
  /**
@@ -611,11 +619,77 @@ export class Actor<Data extends Dictionary = Dictionary> {
  * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
  * @ignore
  */
- async pushData(item: Data | Data[]): Promise<void> {
+ async pushData(item: Data | Data[]): Promise<void>;
+ /**
+ * Stores an object or an array of objects to the default {@apilink Dataset} of the current Actor run.
+ *
+ * This is just a convenient shortcut for {@apilink Dataset.pushData}.
+ * For example, calling the following code:
+ * ```js
+ * await Actor.pushData({ myValue: 123 });
+ * ```
+ *
+ * is equivalent to:
+ * ```js
+ * const dataset = await Actor.openDataset();
+ * await dataset.pushData({ myValue: 123 });
+ * ```
+ *
+ * For more information, see {@apilink Actor.openDataset} and {@apilink Dataset.pushData}
+ *
+ * **IMPORTANT**: Make sure to use the `await` keyword when calling `pushData()`,
+ * otherwise the Actor process might finish before the data are stored!
+ *
+ * @param item Object or array of objects containing data to be stored in the default dataset.
+ * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
+ * @param eventName If provided, the method will attempt to charge for the event for each pushed item.
+ * @ignore
+ */
+ async pushData(item: Data | Data[], eventName: string): Promise<ChargeResult>;
+ /**
+ * Stores an object or an array of objects to the default {@apilink Dataset} of the current Actor run.
+ *
+ * This is just a convenient shortcut for {@apilink Dataset.pushData}.
+ * For example, calling the following code:
+ * ```js
+ * await Actor.pushData({ myValue: 123 });
+ * ```
+ *
+ * is equivalent to:
+ * ```js
+ * const dataset = await Actor.openDataset();
+ * await dataset.pushData({ myValue: 123 });
+ * ```
+ *
+ * For more information, see {@apilink Actor.openDataset} and {@apilink Dataset.pushData}
+ *
+ * **IMPORTANT**: Make sure to use the `await` keyword when calling `pushData()`,
+ * otherwise the Actor process might finish before the data are stored!
+ *
+ * @param item Object or array of objects containing data to be stored in the default dataset.
+ * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
+ * @param eventName If provided, the method will attempt to charge for the event for each pushed item.
+ * @ignore
+ */
+ async pushData(item: Data | Data[], eventName?: string | undefined): Promise<ChargeResult | void> {
  this._ensureActorInit('pushData');
 
  const dataset = await this.openDataset();
- return dataset.pushData(item);
+
+ const maxChargedCount = eventName !== undefined ? this.chargingManager.calculateMaxEventChargeCountWithinLimit(eventName) : Infinity;
+ const toCharge = Array.isArray(item) ? item.length : 1;
+
+ if (toCharge > maxChargedCount) {
+ // Push as many items as we can charge for
+ const items = Array.isArray(item) ? item : [item];
+ await dataset.pushData(items.slice(0, maxChargedCount));
+ } else {
+ await dataset.pushData(item);
+ }
+
+ if (eventName) {
+ return await this.chargingManager.charge({ eventName, count: Math.min(toCharge, maxChargedCount) });
+ }
  }
 
  /**
@@ -896,6 +970,26 @@ export class Actor<Data extends Dictionary = Dictionary> {
  return undefined;
  }
 
+ /**
+ * Charge for a specified number of events - sub-operations of the Actor.
+ *
+ * @param options The name of the event to charge for and the number of events to be charged.
+ * @ignore
+ */
+ async charge(options: ChargeOptions): Promise<ChargeResult> {
+ this._ensureActorInit('charge');
+ return this.chargingManager.charge(options);
+ }
+
+ /**
+ * Retrieve the charging manager to access granular pricing information.
+ * @ignore
+ */
+ getChargingManager(): ChargingManager {
+ this._ensureActorInit('getChargingManager');
+ return this.chargingManager;
+ }
+
  /**
  * Modifies Actor env vars so parsing respects the structure of {@apilink ApifyEnv} interface.
  */
@@ -1305,8 +1399,61 @@ export class Actor<Data extends Dictionary = Dictionary> {
  * @param item Object or array of objects containing data to be stored in the default dataset.
  * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
  */
- static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[]): Promise<void> {
- return Actor.getDefaultInstance().pushData(item);
+ static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[]): Promise<void>;
+ /**
+ * Stores an object or an array of objects to the default {@apilink Dataset} of the current Actor run.
+ *
+ * This is just a convenient shortcut for {@apilink Dataset.pushData}.
+ * For example, calling the following code:
+ * ```js
+ * await Actor.pushData({ myValue: 123 });
+ * ```
+ *
+ * is equivalent to:
+ * ```js
+ * const dataset = await Actor.openDataset();
+ * await dataset.pushData({ myValue: 123 });
+ * ```
+ *
+ * For more information, see {@apilink Actor.openDataset} and {@apilink Dataset.pushData}
+ *
+ * **IMPORTANT**: Make sure to use the `await` keyword when calling `pushData()`,
+ * otherwise the Actor process might finish before the data are stored!
+ *
+ * @param item Object or array of objects containing data to be stored in the default dataset.
+ * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
+ * @param eventName If provided, the method will attempt to charge for the event for each pushed item.
+ */
+ static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[], eventName: string): Promise<ChargeResult>;
+ /**
+ * Stores an object or an array of objects to the default {@apilink Dataset} of the current Actor run.
+ *
+ * This is just a convenient shortcut for {@apilink Dataset.pushData}.
+ * For example, calling the following code:
+ * ```js
+ * await Actor.pushData({ myValue: 123 });
+ * ```
+ *
+ * is equivalent to:
+ * ```js
+ * const dataset = await Actor.openDataset();
+ * await dataset.pushData({ myValue: 123 });
+ * ```
+ *
+ * For more information, see {@apilink Actor.openDataset} and {@apilink Dataset.pushData}
+ *
+ * **IMPORTANT**: Make sure to use the `await` keyword when calling `pushData()`,
+ * otherwise the Actor process might finish before the data are stored!
+ *
+ * @param item Object or array of objects containing data to be stored in the default dataset.
+ * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
+ * @param eventName If provided, the method will attempt to charge for the event for each pushed item.
+ */
+ static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[], eventName?: string): Promise<ChargeResult | void> {
+ if (eventName === undefined) {
+ return await Actor.getDefaultInstance().pushData(item);
+ }
+ return await Actor.getDefaultInstance().pushData(item, eventName);
  }
 
  /**
@@ -1512,6 +1659,22 @@ export class Actor<Data extends Dictionary = Dictionary> {
  return Actor.getDefaultInstance().createProxyConfiguration(proxyConfigurationOptions);
  }
 
+ /**
+ * Charge for a specified number of events - sub-operations of the Actor.
+ *
+ * @param options The name of the event to charge for and the number of events to be charged.
+ */
+ static async charge(options: ChargeOptions): Promise<ChargeResult> {
+ return Actor.getDefaultInstance().charge(options);
+ }
+
+ /**
+ * Retrieve the charging manager to access granular pricing information.
+ */
+ static getChargingManager(): ChargingManager {
+ return Actor.getDefaultInstance().getChargingManager();
+ }
+
  /**
  * Returns a new {@apilink ApifyEnv} object which contains information parsed from all the Apify environment variables.
  *

packages/apify/src/actor.ts

@@ -31,7 +31,10 @@ export interface ConfigurationOptions extends CoreConfigurationOptions {
  userId?: string;
  inputSecretsPrivateKeyPassphrase?: string;
  inputSecretsPrivateKeyFile?: string;
+ maxTotalChargeUsd?: number;
  metaOrigin?: typeof META_ORIGINS[keyof typeof META_ORIGINS];
+ testPayPerEvent?: boolean;
+ useChargingLogDataset?: boolean;
 }
 
 /**
@@ -164,11 +167,14 @@ export class Configuration extends CoreConfiguration {
  ACTOR_TASK_ID: 'actorTaskId',
  ACTOR_WEB_SERVER_PORT: 'containerPort',
  ACTOR_WEB_SERVER_URL: 'containerUrl',
+ ACTOR_MAX_TOTAL_CHARGE_USD: 'maxTotalChargeUsd',
+ ACTOR_TEST_PAY_PER_EVENT: 'testPayPerEvent',
+ ACTOR_USE_CHARGING_LOG_DATASET: 'useChargingLogDataset',
  };
 
  protected static override INTEGER_VARS = [...super.INTEGER_VARS, 'proxyPort', 'containerPort', 'metamorphAfterSleepMillis'];
 
- protected static override BOOLEAN_VARS = [...super.BOOLEAN_VARS, 'isAtHome'];
+ protected static override BOOLEAN_VARS = [...super.BOOLEAN_VARS, 'isAtHome', 'testPayPerEvent', 'useChargingLogDataset'];
 
  protected static override DEFAULTS = {
  ...super.DEFAULTS,
@@ -185,6 +191,8 @@ export class Configuration extends CoreConfiguration {
  standbyPort: +LOCAL_ACTOR_ENV_VARS[ACTOR_ENV_VARS.STANDBY_PORT],
  metamorphAfterSleepMillis: 300e3,
  persistStateIntervalMillis: 60e3, // This value is mentioned in jsdoc in `events.js`, if you update it here, update it there too.
+ testPayPerEvent: false,
+ useChargingLogDataset: false,
  };
 
  /**