feat: support named child runs that survive migration and resurrection

src/actor.ts

@@ -27,6 +27,7 @@ import type {
     ActorStartOptions,
     ApifyClientOptions,
     RunAbortOptions,
+    RunClient,
     RunResurrectOptions,
     TaskCallOptions,
     TaskStartOptions,
@@ -54,6 +55,7 @@ import { ApifyStorageBackend } from './apify_storage_backend.js';
 import type { ChargeOptions, ChargeResult } from './charging.js';
 import { ChargingManager, DEFAULT_DATASET_ITEM_EVENT } from './charging.js';
 import { ChargingStorageBackend } from './charging_storage_backend.js';
+import { ChildRunTracker } from './child_run_tracker.js';
 import type { ConfigurationOptions } from './configuration.js';
 import { Configuration } from './configuration.js';
 import { ActorInputError } from './errors.js';
@@ -327,11 +329,25 @@ export interface Timeout {
     timeout?: number | 'inherit';
 }
 
-export interface CallOptions extends Omit<ActorCallOptions, 'runTimeoutSecs'>, Token, Timeout {}
-export interface StartOptions extends Omit<ActorStartOptions, 'waitForFinish' | 'runTimeoutSecs'>, Token, Timeout {}
-export interface CallTaskOptions extends Omit<TaskCallOptions, 'runTimeoutSecs'>, Token, Timeout {}
+export interface ChildRunOptions {
+    /**
+     * Local name for the child run, unique across all Actors and tasks started by this run.
+     *
+     * If a `READY`, `RUNNING` or `SUCCEEDED` run is already tracked under this name, it is returned instead of
+     * starting a new one, including after a migration or resurrection of this run. A run that was aborted or timed out
+     * is resurrected with the given options, once it finishes aborting or timing out. A run that failed is replaced
+     * by a new one. Reusing the name for a different Actor, task or input throws, unless the tracked run is replaced.
+     */
+    runName?: string;
+}
+
+export interface CallOptions extends Omit<ActorCallOptions, 'runTimeoutSecs'>, Token, Timeout, ChildRunOptions {}
+export interface StartOptions
+    extends Omit<ActorStartOptions, 'waitForFinish' | 'runTimeoutSecs'>, Token, Timeout, ChildRunOptions {}
+export interface CallTaskOptions extends Omit<TaskCallOptions, 'runTimeoutSecs'>, Token, Timeout, ChildRunOptions {}
 export interface ResurrectOptions extends Omit<RunResurrectOptions, 'runTimeoutSecs'>, Token, Timeout {}
-export interface StartTaskOptions extends Omit<TaskStartOptions, 'waitForFinish' | 'runTimeoutSecs'>, Token, Timeout {}
+export interface StartTaskOptions
+    extends Omit<TaskStartOptions, 'waitForFinish' | 'runTimeoutSecs'>, Token, Timeout, ChildRunOptions {}
 
 export interface AbortOptions extends RunAbortOptions, Token {
     /** Exit with given status message */
@@ -451,6 +467,8 @@ export class Actor<Data extends Dictionary = Dictionary> {
     /** @internal */
     static #instance?: Actor;
 
+    #childRunTracker: ChildRunTracker;
+
     /**
      * Configuration of this SDK instance (provided to its constructor). See {@apilink Configuration} for details.
      * @internal
@@ -536,6 +554,7 @@ export class Actor<Data extends Dictionary = Dictionary> {
         this.apifyClient = this.newClient();
         this.eventManager = new PlatformEventManager(this.configuration);
         this.#chargingManager = new ChargingManager(this.configuration, this.apifyClient);
+        this.#childRunTracker = new ChildRunTracker();
     }
 
     /**
@@ -706,6 +725,8 @@ export class Actor<Data extends Dictionary = Dictionary> {
         });
         log.debug(`Default storages purged`);
 
+        await this.#childRunTracker.load();
+
         await this.#chargingManager.init();
         log.debug(`ChargingManager initialized`, this.#chargingManager.getPricingInfo());
 
@@ -866,9 +887,31 @@ export class Actor<Data extends Dictionary = Dictionary> {
      */
     async call(actorId: string, input?: ActorInput, options: CallOptions = {}): Promise<ClientActorRun> {
         const runTimeoutSecs = options.timeout === 'inherit' ? this.getRemainingTimeSecs() : options.timeout;
-        const { token, timeout: _timeout, ...rest } = options;
+        const { token, timeout: _timeout, runName, ...rest } = options;
         const client = token ? this.newClient({ token }) : this.apifyClient;
-        return client.actor(actorId).call(input, { ...rest, runTimeoutSecs });
+
+        if (!runName) return client.actor(actorId).call(input, { ...rest, runTimeoutSecs });
+
+        const { waitSecs, log, timeoutSecs = 'noTimeout', signal, ...otherOptions } = rest;
+        const startOptions = { ...otherOptions, runTimeoutSecs, timeoutSecs, signal };
+        const { run, resumed } = await this.#childRunTracker.startOrResumeChildRun(
+            client,
+            runName,
+            { type: 'actor', id: actorId, input },
+            startOptions,
+            async () => client.actor(actorId).start(input, startOptions),
+        );
+
+        // The earlier part of a resumed run's log was already redirected before the migration.
+        const streamedLog = await client.run(run.id).getStreamedLog({ toLog: log, fromStart: !resumed, signal });
+        streamedLog?.start();
+        try {
+            const finishedRun = await client.run(run.id).waitForFinish({ waitSecs, timeoutSecs, signal });
+            await this.#childRunTracker.update(runName, finishedRun);
+            return finishedRun;
+        } finally {
+            await streamedLog?.stop();
+        }
     }
 
     /**
@@ -896,10 +939,20 @@ export class Actor<Data extends Dictionary = Dictionary> {
      */
     async start(actorId: string, input?: ActorInput, options: StartOptions = {}): Promise<ClientActorRun> {
         const runTimeoutSecs = options.timeout === 'inherit' ? this.getRemainingTimeSecs() : options.timeout;
-        const { token, timeout: _timeout, ...rest } = options;
+        const { token, timeout: _timeout, runName, ...rest } = options;
         const client = token ? this.newClient({ token }) : this.apifyClient;
 
-        return client.actor(actorId).start(input, { ...rest, runTimeoutSecs });
+        if (!runName) return client.actor(actorId).start(input, { ...rest, runTimeoutSecs });
+
+        const startOptions = { ...rest, runTimeoutSecs };
+        const { run } = await this.#childRunTracker.startOrResumeChildRun(
+            client,
+            runName,
+            { type: 'actor', id: actorId, input },
+            startOptions,
+            async () => client.actor(actorId).start(input, startOptions),
+        );
+        return run;
     }
 
     /**
@@ -984,10 +1037,40 @@ export class Actor<Data extends Dictionary = Dictionary> {
      */
     async callTask(taskId: string, input?: Dictionary, options: CallTaskOptions = {}): Promise<ClientActorRun> {
         const runTimeoutSecs = options.timeout === 'inherit' ? this.getRemainingTimeSecs() : options.timeout;
-        const { token, timeout: _timeout, ...rest } = options;
+        const { token, timeout: _timeout, runName, ...rest } = options;
         const client = token ? this.newClient({ token }) : this.apifyClient;
 
-        return client.task(taskId).call(input, { ...rest, runTimeoutSecs });
+        if (!runName) return client.task(taskId).call(input, { ...rest, runTimeoutSecs });
+
+        const { waitSecs, timeoutSecs = 'noTimeout', signal, ...otherOptions } = rest;
+        const startOptions = { ...otherOptions, runTimeoutSecs, timeoutSecs, signal };
+        const { run } = await this.#childRunTracker.startOrResumeChildRun(
+            client,
+            runName,
+            { type: 'task', id: taskId, input },
+            startOptions,
+            async () => client.task(taskId).start(input, startOptions),
+        );
+        const finishedRun = await client.run(run.id).waitForFinish({ waitSecs, timeoutSecs, signal });
+        await this.#childRunTracker.update(runName, finishedRun);
+        return finishedRun;
+    }
+
+    /**
+     * Clients for the child runs started with the `runName` option, keyed by that name, including the runs started
+     * before a migration or resurrection of this run. Each client points to the current run under its name.
+     *
+     * ```js
+     * await Actor.childRuns['my-child'].waitForFinish();
+     * ```
+     *
+     * A run started with a custom `token` uses that token, except for a run started before a migration or
+     * resurrection, which uses the default client. The tracked runs, including the earlier runs replaced
+     * under the same name, are stored in the default key-value store under the `__ACTOR_CHILD_RUNS` key.
+     */
+    get childRuns(): Record<string, RunClient> {
+        this.ensureActorInit('childRuns');
+        return this.#childRunTracker.getRunClients(this.apifyClient);
     }
 
     /**
@@ -1015,10 +1098,20 @@ export class Actor<Data extends Dictionary = Dictionary> {
      */
     async startTask(taskId: string, input?: Dictionary, options: StartTaskOptions = {}): Promise<ClientActorRun> {
         const runTimeoutSecs = options.timeout === 'inherit' ? this.getRemainingTimeSecs() : options.timeout;
-        const { token, timeout: _timeout, ...rest } = options;
+        const { token, timeout: _timeout, runName, ...rest } = options;
         const client = token ? this.newClient({ token }) : this.apifyClient;
 
-        return client.task(taskId).start(input, { ...rest, runTimeoutSecs });
+        if (!runName) return client.task(taskId).start(input, { ...rest, runTimeoutSecs });
+
+        const startOptions = { ...rest, runTimeoutSecs };
+        const { run } = await this.#childRunTracker.startOrResumeChildRun(
+            client,
+            runName,
+            { type: 'task', id: taskId, input },
+            startOptions,
+            async () => client.task(taskId).start(input, startOptions),
+        );
+        return run;
     }
 
     /**
@@ -2028,6 +2121,22 @@ export class Actor<Data extends Dictionary = Dictionary> {
         return Actor.getDefaultInstance().start(actorId, input, options);
     }
 
+    /**
+     * Clients for the child runs started with the `runName` option, keyed by that name, including the runs started
+     * before a migration or resurrection of this run. Each client points to the current run under its name.
+     *
+     * ```js
+     * await Actor.childRuns['my-child'].waitForFinish();
+     * ```
+     *
+     * A run started with a custom `token` uses that token, except for a run started before a migration or
+     * resurrection, which uses the default client. The tracked runs, including the earlier runs replaced
+     * under the same name, are stored in the default key-value store under the `__ACTOR_CHILD_RUNS` key.
+     */
+    static get childRuns(): Record<string, RunClient> {
+        return Actor.getDefaultInstance().childRuns;
+    }
+
     /**
      * Aborts given Actor run on the Apify platform using the current user account (determined by the `APIFY_TOKEN` environment variable).
      *

src/charging.ts

@@ -1,11 +1,10 @@
-import { AsyncLocalStorage } from 'node:async_hooks';
-
 import { Dataset, KeyValueStore } from '@crawlee/core';
 import type { ActorRunPricingInfo, ApifyClient } from 'apify-client';
 
 import log from '@apify/log';
 
 import type { Configuration } from './configuration.js';
+import { ReentrantAsyncLock } from './utils.js';
 
 interface ChargingStateItem {
     chargeCount: number;
@@ -74,26 +73,6 @@ export interface ActorPricingInfo {
     perEventPrices: Record<string, number>;
 }
 
-/**
- * A FIFO mutex that a critical section may re-enter from a nested call — the charge lock is taken by
- * `Actor.pushData()` and again, one level down, by the dataset backend it pushes through.
- */
-class ReentrantAsyncLock {
-    #tail: Promise<unknown> = Promise.resolve();
-    readonly #held = new AsyncLocalStorage<true>();
-
-    async runExclusive<T>(fn: () => Promise<T>): Promise<T> {
-        if (this.#held.getStore()) {
-            return await fn();
-        }
-
-        const run = this.#tail.then(async () => this.#held.run(true, fn));
-        // Keep the chain alive even when the critical section throws.
-        this.#tail = run.catch(() => {});
-        return await run;
-    }
-}
-
 /**
  * Handles pay-per-event charging.
  */