diff --git a/src/actor.ts b/src/actor.ts index cad947ef00..f39d8a1fc5 100644 --- a/src/actor.ts +++ b/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, Token, Timeout {} -export interface StartOptions extends Omit, Token, Timeout {} -export interface CallTaskOptions extends Omit, 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, Token, Timeout, ChildRunOptions {} +export interface StartOptions + extends Omit, Token, Timeout, ChildRunOptions {} +export interface CallTaskOptions extends Omit, Token, Timeout, ChildRunOptions {} export interface ResurrectOptions extends Omit, Token, Timeout {} -export interface StartTaskOptions extends Omit, Token, Timeout {} +export interface StartTaskOptions + extends Omit, Token, Timeout, ChildRunOptions {} export interface AbortOptions extends RunAbortOptions, Token { /** Exit with given status message */ @@ -451,6 +467,8 @@ export class Actor { /** @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 { 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 { }); 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 { */ async call(actorId: string, input?: ActorInput, options: CallOptions = {}): Promise { 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 { */ async start(actorId: string, input?: ActorInput, options: StartOptions = {}): Promise { 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 { */ async callTask(taskId: string, input?: Dictionary, options: CallTaskOptions = {}): Promise { 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 { + this.ensureActorInit('childRuns'); + return this.#childRunTracker.getRunClients(this.apifyClient); } /** @@ -1015,10 +1098,20 @@ export class Actor { */ async startTask(taskId: string, input?: Dictionary, options: StartTaskOptions = {}): Promise { 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 { 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 { + return Actor.getDefaultInstance().childRuns; + } + /** * Aborts given Actor run on the Apify platform using the current user account (determined by the `APIFY_TOKEN` environment variable). * diff --git a/src/charging.ts b/src/charging.ts index eaa529ff98..53036363f4 100644 --- a/src/charging.ts +++ b/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; } -/** - * 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 = Promise.resolve(); - readonly #held = new AsyncLocalStorage(); - - async runExclusive(fn: () => Promise): Promise { - 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. */ diff --git a/src/child_run_tracker.ts b/src/child_run_tracker.ts new file mode 100644 index 0000000000..b0d06c4ede --- /dev/null +++ b/src/child_run_tracker.ts @@ -0,0 +1,262 @@ +import { createHash } from 'node:crypto'; + +import { KeyValueStore, withDirectStorageAccess } from '@crawlee/core'; +import { sleep } from '@crawlee/utils'; +import type { Run as ActorRun, RunClient, RunResurrectOptions } from 'apify-client'; +import { z } from 'zod'; + +import { ReentrantAsyncLock } from './utils.js'; +import type { ApifyClient } from './index.js'; + +/** + * Status of a tracked child run. This copies the `ActorRun['status']` type, + * but adds `LOST` to mark a locally tracked run that the platform no longer returns. + * + * `LOST` runs can happen as a result of a manual removal or expiration. + */ +export type ChildRunStatus = ActorRun['status'] | 'LOST'; + +export interface ChildRunInfo { + runId: string; + status: ChildRunStatus; + /** ISO 8601 timestamp. */ + startedAt: string; +} + +export interface TrackedChildRun extends ChildRunInfo { + /** Identifies the Actor / task and input the run was started with. */ + checksum: string; + /** Earlier runs that were replaced under the same name, oldest first. */ + history: ChildRunInfo[]; +} + +export interface ChildRunRequest { + type: 'actor' | 'task'; + id: string; + input?: unknown; +} + +const sortKeys = (value: unknown): unknown => { + if (Array.isArray(value)) return value.map(sortKeys); + if (value === null || typeof value !== 'object') return value; + + return Object.fromEntries( + Object.entries(value) + .sort(([a], [b]) => Number(a > b) - Number(a < b)) + .map(([key, nested]) => [key, sortKeys(nested)]), + ); +}; + +// Hashes the JSON form that `apify-client` sends (`Date` -> ISO string, `URL` -> href, functions -> source), +// not the objects' own keys. +const toSentJson = (value: unknown) => + JSON.parse(JSON.stringify(value, (_key, nested) => (typeof nested === 'function' ? nested.toString() : nested))); + +const checksumRequest = ({ type, id, input }: ChildRunRequest) => + createHash('sha256') + .update(JSON.stringify(sortKeys(toSentJson({ type, id, input })))) + .digest('hex'); + +const toInfo = (run: ActorRun): ChildRunInfo => ({ + runId: run.id, + status: run.status, + startedAt: run.startedAt.toISOString(), +}); + +const CHILD_RUNS_KVS_KEY = '__ACTOR_CHILD_RUNS'; + +const childRunInfoSchema = z.object({ runId: z.string(), status: z.string(), startedAt: z.string() }); +const trackedRunsSchema = z.record( + z.string(), + childRunInfoSchema.extend({ checksum: z.string(), history: z.array(childRunInfoSchema).default([]) }), +); + +/** How many times a tracked run that the API reports as missing is looked up again before it counts as gone. */ +const NOT_FOUND_RETRIES = 12; +const NOT_FOUND_RETRY_INTERVAL_MILLIS = 250; + +/** + * Fetches a tracked run, retrying a 404 for a few seconds. A run started moments ago (e.g. by a concurrent call or + * right before a migration) may not be on every API replica yet, and treating it as gone would start a duplicate. + */ +const getTrackedRun = async (runClient: RunClient) => { + let run = await runClient.get(); + for (let retry = 0; !run && retry < NOT_FOUND_RETRIES; retry++) { + await sleep(NOT_FOUND_RETRY_INTERVAL_MILLIS); + run = await runClient.get(); + } + return run; +}; + +/** Statuses that shortly end as `ABORTED` / `TIMED-OUT`. */ +const SETTLING_STATUSES = ['ABORTING', 'TIMING-OUT']; +const RESURRECTABLE_STATUSES = ['ABORTED', 'TIMED-OUT']; +const REUSABLE_STATUSES = ['READY', 'RUNNING', 'SUCCEEDED', ...RESURRECTABLE_STATUSES]; + +export class ChildRunTracker { + private trackedRuns: Record = {}; + private loading?: Promise; + /** Client each name was last started or resumed with in this process, so a custom `token` is kept. */ + private clients = new Map(); + private lastWrite: Promise = Promise.resolve(); + private locks = new Map(); + + /** + * Throws if the run tracked under `runName` was started for a different Actor / task or input, + * as it would otherwise be silently returned or resurrected in place of the requested one. + */ + async verifyRequest(runName: string, request: ChildRunRequest) { + const tracked = await this.get(runName); + + if (tracked && tracked.checksum !== checksumRequest(request)) { + throw new Error( + `The run name "${runName}" was already used for a different Actor, task or input. Use a unique \`runName\` for each child run.`, + ); + } + } + + /** + * Returns the child run tracked under `runName` if it is still in progress or has succeeded, and resurrects it + * with `startOptions` if it was aborted or timed out (waiting for it to finish aborting or timing out first). + * Otherwise (the run failed or is gone), starts a new one using `start` and tracks it under `runName`, + * keeping the previous one in the history. + */ + async startOrResumeChildRun( + client: ApifyClient, + runName: string, + request: ChildRunRequest, + startOptions: RunResurrectOptions, + start: () => Promise, + ): Promise<{ run: ActorRun; resumed: boolean }> { + if (!this.locks.has(runName)) { + this.locks.set(runName, new ReentrantAsyncLock()); + } + + return this.locks.get(runName)!.runExclusive(async () => { + const { build, memory, runTimeoutSecs, maxItems, maxTotalChargeUsd, restartOnError, timeoutSecs, signal } = + startOptions; + const tracked = await this.get(runName); + let trackedRun = tracked ? await getTrackedRun(client.run(tracked.runId)) : undefined; + + if (trackedRun && SETTLING_STATUSES.includes(trackedRun.status)) { + trackedRun = await client.run(trackedRun.id).waitForFinish({ timeoutSecs, signal }); + } + + if (trackedRun && REUSABLE_STATUSES.includes(trackedRun.status)) { + await this.verifyRequest(runName, request); + const run = RESURRECTABLE_STATUSES.includes(trackedRun.status) + ? await client.run(trackedRun.id).resurrect({ + build, + memory, + runTimeoutSecs, + maxItems, + maxTotalChargeUsd, + restartOnError, + timeoutSecs, + signal, + }) + : trackedRun; + await this.update(runName, run); + this.clients.set(runName, client); + return { run, resumed: true }; + } + + const run = await start(); + await this.track(runName, run, request, trackedRun?.status ?? 'LOST'); + this.clients.set(runName, client); + return { run, resumed: false }; + }); + } + + /** + * Tracks `run` as the current run under `runName`. A run tracked earlier under the same name is moved to the + * history, with `replacedStatus` as its last known status. + */ + async track(runName: string, run: ActorRun, request: ChildRunRequest, replacedStatus: ChildRunStatus) { + const trackedRuns = await this.load(); + const previous = trackedRuns[runName]; + + trackedRuns[runName] = { + ...toInfo(run), + checksum: checksumRequest(request), + history: previous + ? [ + ...previous.history, + { runId: previous.runId, startedAt: previous.startedAt, status: replacedStatus }, + ] + : [], + }; + + await this.persist(); + } + + /** + * Records the latest observed state of the run tracked under `runName`. + */ + async update(runName: string, run: ActorRun) { + const trackedRuns = await this.load(); + const tracked = trackedRuns[runName]; + + if (tracked?.runId !== run.id || tracked.status === run.status) return; + + tracked.status = run.status; + await this.persist(); + } + + /** + * Returns the tracked run and its history for `runName`, if there is one. + */ + async get(runName: string) { + const trackedRuns = await this.load(); + return trackedRuns[runName]; + } + + /** + * Returns a `RunClient` for the current run under each name loaded so far. The client is the one the name was + * last started or resumed with in this process, or `defaultClient` (e.g. for a run tracked before a migration). + */ + getRunClients(defaultClient: ApifyClient): Record { + return Object.fromEntries( + Object.entries(this.trackedRuns).map(([runName, { runId }]) => [ + runName, + (this.clients.get(runName) ?? defaultClient).run(runId), + ]), + ); + } + + private async persist() { + const write = this.lastWrite.then(async () => + withDirectStorageAccess(async () => { + const defaultStore = await KeyValueStore.open(); + await defaultStore.setValue(CHILD_RUNS_KVS_KEY, this.trackedRuns); + }), + ); + this.lastWrite = write.catch(() => {}); + await write; + } + + /** + * Loads the tracked runs from the default key-value store, once. + */ + async load() { + this.loading ??= KeyValueStore.open() + .then(async (defaultStore) => defaultStore.getValue(CHILD_RUNS_KVS_KEY)) + .then((storedRuns) => { + const parsed = trackedRunsSchema.safeParse(storedRuns ?? {}); + if (!parsed.success) { + throw new Error( + `The child runs stored under the "${CHILD_RUNS_KVS_KEY}" key in the default key-value store are malformed.`, + { cause: parsed.error }, + ); + } + this.trackedRuns = parsed.data as Record; + }) + .catch((error) => { + this.loading = undefined; + throw error; + }); + + await this.loading; + return this.trackedRuns; + } +} diff --git a/src/utils.ts b/src/utils.ts index c085103004..4d20e417c0 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -1,3 +1,4 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; import { existsSync, readFileSync } from 'node:fs'; import { createRequire } from 'node:module'; import { type } from 'node:os'; @@ -138,3 +139,22 @@ export function printOutdatedSdkWarning() { log.warning(`You are using an outdated version (${apifyPkgJson.version}) of Apify SDK. We recommend you to update to the latest version (${latestApifyVersion}). Read more about Apify SDK versioning at: https://help.apify.com/en/articles/3184510-updates-and-versioning-of-apify-sdk`); } + +/** + * A FIFO mutex that a critical section may re-enter from a nested call without deadlocking. + */ +export class ReentrantAsyncLock { + #tail: Promise = Promise.resolve(); + readonly #held = new AsyncLocalStorage(); + + async runExclusive(fn: () => Promise): Promise { + 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; + } +} diff --git a/test/apify/charging.test.ts b/test/apify/charging.test.ts index 22bc091778..3e921b2f3b 100644 --- a/test/apify/charging.test.ts +++ b/test/apify/charging.test.ts @@ -5,7 +5,8 @@ import { withStorageTransaction, } from '@crawlee/core'; import { Actor } from 'apify'; -import type { RunClient } from 'apify-client'; +import type { KeyValueStoreResource, RunClient } from 'apify-client'; +import { KeyValueStoreClient } from 'apify-client'; import type { MockInstance } from 'vitest'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; @@ -53,6 +54,10 @@ function setUpPlatformEnv( } process.env.APIFY_CHARGED_ACTOR_EVENT_COUNTS = JSON.stringify(options.chargedEventCounts ?? {}); + + // `Actor.init()` reads the tracked child runs from the default key-value store. + vitest.spyOn(KeyValueStoreClient.prototype, 'get').mockResolvedValue({ id: 'default' } as KeyValueStoreResource); + vitest.spyOn(KeyValueStoreClient.prototype, 'getRecord').mockResolvedValue(undefined); } /** diff --git a/test/apify/child_runs.test.ts b/test/apify/child_runs.test.ts new file mode 100644 index 0000000000..f1433467a3 --- /dev/null +++ b/test/apify/child_runs.test.ts @@ -0,0 +1,453 @@ +import { MemoryStorageBackend } from '@crawlee/core'; +import { sleep } from '@crawlee/utils'; +import type { Actor } from 'apify'; +import type { Run as ActorRun } from 'apify-client'; +import { ActorClient, RunClient, TaskClient } from 'apify-client'; +import type { MockInstance } from 'vitest'; +import { onTestFinished } from 'vitest'; + +import { ACTOR_ENV_VARS, APIFY_ENV_VARS } from '@apify/consts'; +import log from '@apify/log'; + +import type { TrackedChildRun } from '../../src/child_run_tracker.js'; +import { createIsolatedActor } from '../createIsolatedActor.js'; + +// Skips the delays between the lookups of a tracked run that the API reports as missing. +vitest.mock('@crawlee/utils', async (importOriginal) => ({ + ...(await importOriginal>()), + sleep: vitest.fn(async () => {}), +})); + +const globalOptions = { + token: 'some-token', + actId: 'some-act-id', + defaultKeyValueStoreId: 'some-store-id', + input: { foo: 'bar' }, + contentType: 'application/json', + outputKey: 'OUTPUT', + outputValue: 'some-output', + build: 'xxx', + taskId: 'some-task-id', + runId: 'some-run-id', + targetActorId: 'some-target-actor-id', +}; + +describe('child run tracking with `runName`', () => { + const { input, actId, taskId } = globalOptions; + const runName = 'child'; + const startedAt = new Date('2026-01-01T00:00:00.000Z'); + const startedRun = { id: 'child-run', status: 'RUNNING', startedAt } as ActorRun; + const finishedRun = { ...startedRun, status: 'SUCCEEDED' } as ActorRun; + const trackedInfo = { runId: startedRun.id, status: 'RUNNING', startedAt: startedAt.toISOString() }; + const stored = { ...trackedInfo, checksum: expect.any(String) }; + + let storage: MemoryStorageBackend; + let startSpy: MockInstance; + let taskStartSpy: MockInstance; + let getSpy: MockInstance; + let waitForFinishSpy: MockInstance; + let getStreamedLogSpy: MockInstance; + let resurrectSpy: MockInstance; + + const getChildRunLedger = async (actor: Actor) => + (await actor.openKeyValueStore()).getValue>('__ACTOR_CHILD_RUNS'); + const newActor = () => createIsolatedActor({ storageClient: storage }).actor; + + beforeEach(() => { + storage = new MemoryStorageBackend(); + startSpy = vitest.spyOn(ActorClient.prototype, 'start').mockResolvedValue(startedRun); + taskStartSpy = vitest.spyOn(TaskClient.prototype, 'start').mockResolvedValue(startedRun); + getSpy = vitest.spyOn(RunClient.prototype, 'get').mockResolvedValue(startedRun); + waitForFinishSpy = vitest.spyOn(RunClient.prototype, 'waitForFinish').mockResolvedValue(finishedRun); + getStreamedLogSpy = vitest.spyOn(RunClient.prototype, 'getStreamedLog').mockResolvedValue(undefined); + resurrectSpy = vitest.spyOn(RunClient.prototype, 'resurrect').mockResolvedValue(startedRun); + }); + + test('start() without a run name does not track the run', async () => { + const actor = newActor(); + + await actor.start(actId, input); + + expect(startSpy).toBeCalledTimes(1); + expect(await getChildRunLedger(actor)).toBeNull(); + }); + + test('start() starts a new run and tracks it', async () => { + const actor = newActor(); + + const run = await actor.start(actId, input, { runName }); + + expect(run).toEqual(startedRun); + expect(startSpy).toBeCalledTimes(1); + expect(await getChildRunLedger(actor)).toEqual({ + [runName]: { ...stored, history: [] }, + }); + }); + + test.each(['READY', 'RUNNING', 'SUCCEEDED'] as const)( + 'start() respects an existing Actor run with %s status', + async (status) => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + getSpy.mockResolvedValue({ ...startedRun, status }); + + const actor = newActor(); + const run = await actor.start(actId, input, { runName }); + + expect(startSpy).not.toBeCalled(); + expect(run).toEqual({ ...startedRun, status }); + expect(getSpy).toBeCalledTimes(1); + expect(await getChildRunLedger(actor)).toEqual({ [runName]: { ...stored, status, history: [] } }); + }, + ); + + test('start() starts a new run when the tracked run failed and keeps the old one in the history', async () => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + getSpy.mockResolvedValue({ ...startedRun, status: 'FAILED' }); + const replacement = { id: 'replacement-run', status: 'RUNNING', startedAt } as ActorRun; + startSpy.mockResolvedValue(replacement); + + const actor = newActor(); + const run = await actor.start(actId, input, { runName }); + + expect(startSpy).toBeCalledTimes(1); + expect(resurrectSpy).not.toBeCalled(); + expect(run).toEqual(replacement); + expect(await getChildRunLedger(actor)).toEqual({ + [runName]: { + ...stored, + runId: replacement.id, + history: [{ ...trackedInfo, status: 'FAILED' }], + }, + }); + }); + + test.each(['ABORTED', 'TIMED-OUT'] as const)( + 'start() resurrects a tracked run in the %s status with the resurrect options of the call', + async (status) => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + getSpy.mockResolvedValue({ ...startedRun, status }); + + const actor = newActor(); + const run = await actor.start(actId, input, { + runName, + build: 'beta', + memory: 1024, + timeout: 60, + maxItems: 10, + webhooks: [], + }); + + expect(startSpy).not.toBeCalled(); + expect(resurrectSpy).toBeCalledWith({ build: 'beta', memory: 1024, runTimeoutSecs: 60, maxItems: 10 }); + expect(run).toEqual(startedRun); + expect(await getChildRunLedger(actor)).toEqual({ [runName]: { ...stored, history: [] } }); + }, + ); + + test.each([ + ['ABORTING', 'ABORTED'], + ['TIMING-OUT', 'TIMED-OUT'], + ] as const)( + 'start() waits for a tracked run in the %s status to finish before resurrecting it', + async (status, finalStatus) => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + getSpy.mockResolvedValue({ ...startedRun, status }); + waitForFinishSpy.mockResolvedValue({ ...startedRun, status: finalStatus }); + + await newActor().start(actId, input, { runName }); + + expect(waitForFinishSpy).toBeCalledTimes(1); + expect(resurrectSpy).toBeCalledTimes(1); + expect(startSpy).not.toBeCalled(); + }, + ); + + test("start() resurrects a tracked run with the remaining time of this run with `timeout: 'inherit'`", async () => { + await newActor().start(actId, input, { runName }); + getSpy.mockResolvedValue({ ...startedRun, status: 'TIMED-OUT' }); + vitest.setSystemTime(startedAt); + process.env[APIFY_ENV_VARS.IS_AT_HOME] = '1'; + process.env[ACTOR_ENV_VARS.TIMEOUT_AT] = new Date(startedAt.getTime() + 60_000).toISOString(); + onTestFinished(() => { + delete process.env[APIFY_ENV_VARS.IS_AT_HOME]; + delete process.env[ACTOR_ENV_VARS.TIMEOUT_AT]; + vitest.useRealTimers(); + }); + + await newActor().start(actId, input, { runName, timeout: 'inherit' }); + + expect(resurrectSpy).toBeCalledWith(expect.objectContaining({ runTimeoutSecs: 60 })); + }); + + test('start() starts a new run when the tracked run no longer exists and records the old one as lost', async () => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + getSpy.mockResolvedValue(undefined); + startSpy.mockResolvedValue({ ...startedRun, id: 'replacement-run' }); + + const actor = newActor(); + await actor.start(actId, input, { runName }); + + expect(startSpy).toBeCalledTimes(1); + expect(getSpy).toBeCalledTimes(13); + expect(sleep).toBeCalledWith(250); + expect((await getChildRunLedger(actor))![runName].history).toEqual([{ ...trackedInfo, status: 'LOST' }]); + }); + + test('start() looks a missing tracked run up again, as it may not have reached every API replica yet', async () => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + getSpy.mockResolvedValueOnce(undefined).mockResolvedValueOnce(undefined); + + const run = await newActor().start(actId, input, { runName }); + + expect(run).toEqual(startedRun); + expect(getSpy).toBeCalledTimes(3); + expect(startSpy).not.toBeCalled(); + }); + + test('repeated replacements accumulate in the history, oldest first', async () => { + const actor = newActor(); + await actor.start(actId, input, { runName }); + getSpy.mockResolvedValue({ ...startedRun, status: 'FAILED' }); + startSpy.mockResolvedValueOnce({ ...startedRun, id: 'second' }); + await actor.start(actId, input, { runName }); + getSpy.mockResolvedValue({ ...startedRun, id: 'second', status: 'FAILED' }); + startSpy.mockResolvedValueOnce({ ...startedRun, id: 'third' }); + await actor.start(actId, input, { runName }); + + const tracked = (await getChildRunLedger(actor))![runName]; + expect(tracked.runId).toBe('third'); + expect(tracked.history.map(({ runId, status }) => [runId, status])).toEqual([ + [startedRun.id, 'FAILED'], + ['second', 'FAILED'], + ]); + }); + + test('start() tracks different run names independently', async () => { + const actor = newActor(); + startSpy.mockResolvedValueOnce({ ...startedRun, id: 'first' }).mockResolvedValueOnce({ + ...startedRun, + id: 'second', + }); + + await Promise.all([actor.start(actId, input, { runName: 'a' }), actor.start(actId, input, { runName: 'b' })]); + + expect(Object.fromEntries(Object.entries(actor.childRuns).map(([name, { id }]) => [name, id]))).toEqual({ + a: 'first', + b: 'second', + }); + }); + + test('childRuns returns a client for the current run under each name', async () => { + const actor = newActor(); + await actor.start(actId, input, { runName }); + getSpy.mockResolvedValue({ ...startedRun, status: 'FAILED' }); + startSpy.mockResolvedValueOnce({ ...startedRun, id: 'replacement-run' }); + await actor.start(actId, input, { runName }); + + const runClient = actor.childRuns[runName]; + + expect(runClient.id).toBe('replacement-run'); + expect(await runClient.waitForFinish()).toEqual(finishedRun); + }); + + test('init() loads the child runs tracked before a migration', async () => { + await newActor().start(actId, input, { runName }); + + const actor = newActor(); + await actor.init(); + + expect(Object.keys(actor.childRuns)).toEqual([runName]); + expect(actor.childRuns[runName].id).toBe(startedRun.id); + }); + + test('init() throws when the stored child runs are malformed', async () => { + await (await newActor().openKeyValueStore()).setValue('__ACTOR_CHILD_RUNS', { [runName]: { runId: 42 } }); + + await expect(newActor().init()).rejects.toThrow(/__ACTOR_CHILD_RUNS.*malformed/); + }); + + test('childRuns uses the token a run was started with in this process', async () => { + await newActor().start(actId, input, { runName: 'before-migration' }); + + const actor = newActor(); + await actor.init(); + await actor.start(actId, input, { runName: 'custom-token', token: 'custom-token' }); + + expect(actor.childRuns['before-migration'].apifyClient).toBe(actor.apifyClient); + expect(actor.childRuns['custom-token'].apifyClient.token).toBe('custom-token'); + }); + + test('childRuns warns when accessed before init()', () => { + const warningSpy = vitest.spyOn(log, 'warning'); + + expect(newActor().childRuns).toEqual({}); + expect(warningSpy).toHaveBeenCalledWith(expect.stringContaining('Actor.init()')); + }); + + test('reusing a run name for a different Actor in the same process throws', async () => { + const actor = newActor(); + await actor.start(actId, input, { runName }); + + await expect(actor.start('other-actor', input, { runName })).rejects.toThrow(/already used/); + await expect(actor.callTask(taskId, input, { runName })).rejects.toThrow(/already used/); + expect(startSpy).toBeCalledTimes(1); + }); + + test('reusing a run name with a different input in the same process throws', async () => { + const actor = newActor(); + await actor.start(actId, input, { runName }); + + await expect(actor.start(actId, { foo: 'baz' }, { runName })).rejects.toThrow(/already used/); + }); + + test('reusing a run name with an equivalent input in the same process deduplicates the run', async () => { + const actor = newActor(); + await actor.start(actId, { a: 1, b: { c: 2, d: 3 } }, { runName }); + + await actor.start(actId, { b: { d: 3, c: 2 }, a: 1 }, { runName }); + + expect(startSpy).toBeCalledTimes(1); + }); + + test('reusing a run name with a different input after a migration throws', async () => { + await newActor().start(actId, input, { runName }); + + await expect(newActor().start(actId, { foo: 'baz' }, { runName })).rejects.toThrow(/already used/); + expect(startSpy).toBeCalledTimes(1); + }); + + test('reusing a run name with the same input after a migration deduplicates the run', async () => { + await newActor().start(actId, { a: 1, b: 2 }, { runName }); + + await newActor().start(actId, { b: 2, a: 1 }, { runName }); + + expect(startSpy).toBeCalledTimes(1); + }); + + test('reusing a run name with a different input throws when the tracked run would be resurrected', async () => { + const actor = newActor(); + await actor.start(actId, input, { runName }); + getSpy.mockResolvedValue({ ...startedRun, status: 'ABORTED' }); + + await expect(actor.start(actId, { foo: 'baz' }, { runName })).rejects.toThrow(/already used/); + expect(resurrectSpy).not.toBeCalled(); + }); + + test('reusing a run name with a different input is allowed when the tracked run is replaced', async () => { + const actor = newActor(); + await actor.start(actId, input, { runName }); + getSpy.mockResolvedValue({ ...startedRun, status: 'FAILED' }); + + await actor.start(actId, { foo: 'baz' }, { runName }); + + expect(startSpy).toBeCalledTimes(2); + }); + + test('call() waits for a new run started with a run name', async () => { + const actor = newActor(); + + const run = await actor.call(actId, input, { runName, waitSecs: 10 }); + + expect(run).toEqual(finishedRun); + expect(startSpy).toBeCalledWith(input, expect.objectContaining({ timeoutSecs: 'noTimeout' })); + expect(waitForFinishSpy).toBeCalledWith({ waitSecs: 10, timeoutSecs: 'noTimeout' }); + expect(getStreamedLogSpy).toBeCalledWith({ toLog: undefined, fromStart: true }); + }); + + test('call() with a run name forwards the signal and request timeout like `ActorClient.call`', async () => { + const { signal } = new AbortController(); + + await newActor().call(actId, input, { runName, signal, timeoutSecs: 30 }); + + expect(startSpy).toBeCalledWith(input, expect.objectContaining({ signal, timeoutSecs: 30 })); + expect(getStreamedLogSpy).toBeCalledWith(expect.objectContaining({ signal })); + expect(waitForFinishSpy).toBeCalledWith({ timeoutSecs: 30, signal }); + }); + + test('call() waits for the tracked run instead of starting a new one', async () => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + + const actor = newActor(); + const run = await actor.call(actId, input, { runName }); + + expect(run).toEqual(finishedRun); + expect(startSpy).not.toBeCalled(); + expect(getStreamedLogSpy).toBeCalledWith({ toLog: undefined, fromStart: false }); + expect((await getChildRunLedger(actor))![runName].status).toBe('SUCCEEDED'); + }); + + test('call() waits for a resurrected run without replaying its log', async () => { + await newActor().start(actId, input, { runName }); + startSpy.mockClear(); + getSpy.mockResolvedValue({ ...startedRun, status: 'ABORTED' }); + + const run = await newActor().call(actId, input, { runName }); + + expect(run).toEqual(finishedRun); + expect(startSpy).not.toBeCalled(); + expect(resurrectSpy).toBeCalledWith(expect.objectContaining({ timeoutSecs: 'noTimeout' })); + expect(getStreamedLogSpy).toBeCalledWith(expect.objectContaining({ fromStart: false })); + }); + + test('call() stops the log stream when waiting for the run fails', async () => { + const streamedLog = { start: vitest.fn(), stop: vitest.fn().mockResolvedValue(undefined) }; + getStreamedLogSpy.mockResolvedValue(streamedLog as any); + waitForFinishSpy.mockRejectedValue(new Error('boom')); + + await expect(newActor().call(actId, input, { runName })).rejects.toThrow('boom'); + + expect(streamedLog.start).toBeCalledTimes(1); + expect(streamedLog.stop).toBeCalledTimes(1); + }); + + test('call() without a run name uses `ActorClient.call`', async () => { + const callSpy = vitest.spyOn(ActorClient.prototype, 'call').mockResolvedValue(finishedRun); + + await newActor().call(actId, input); + + expect(callSpy).toBeCalledTimes(1); + expect(startSpy).not.toBeCalled(); + }); + + test('startTask() returns the tracked run instead of starting a new one', async () => { + await newActor().startTask(taskId, input, { runName }); + taskStartSpy.mockClear(); + + const actor = newActor(); + const run = await actor.startTask(taskId, input, { runName }); + + expect(run).toEqual(startedRun); + expect(taskStartSpy).not.toBeCalled(); + expect(await getChildRunLedger(actor)).toEqual({ [runName]: { ...stored, history: [] } }); + }); + + test('callTask() deduplicates the tracked run instead of starting a new one', async () => { + await newActor().callTask(taskId, input, { runName }); + expect(taskStartSpy).toBeCalledTimes(1); + taskStartSpy.mockClear(); + + const actor = newActor(); + const run = await actor.callTask(taskId, input, { runName, waitSecs: 5 }); + + expect(run).toEqual(finishedRun); + expect(taskStartSpy).not.toBeCalled(); + expect(waitForFinishSpy).toHaveBeenLastCalledWith({ waitSecs: 5, timeoutSecs: 'noTimeout' }); + expect((await getChildRunLedger(actor))![runName].status).toBe('SUCCEEDED'); + }); + + test('callTask() with a run name forwards the signal and request timeout like `TaskClient.call`', async () => { + const { signal } = new AbortController(); + + await newActor().callTask(taskId, input, { runName, signal, timeoutSecs: 30 }); + + expect(taskStartSpy).toBeCalledWith(input, expect.objectContaining({ signal, timeoutSecs: 30 })); + expect(waitForFinishSpy).toBeCalledWith({ timeoutSecs: 30, signal }); + }); +}); diff --git a/test/e2e/sdk/actorChildRuns/src/main.mjs b/test/e2e/sdk/actorChildRuns/src/main.mjs new file mode 100644 index 0000000000..b69c650f63 --- /dev/null +++ b/test/e2e/sdk/actorChildRuns/src/main.mjs @@ -0,0 +1,47 @@ +import assert from 'node:assert/strict'; +import { setTimeout } from 'node:timers/promises'; + +import { Actor } from 'apify'; + +await Actor.init(); + +const { mode, childSleepSecs } = (await Actor.getInput()) ?? {}; +const { actorId } = Actor.getEnv(); +const childInput = { childSleepSecs: mode === 'resurrect' ? 300 : 20 }; +const childRunId = await Actor.getValue('CHILD_RUN_ID'); + +if (childSleepSecs !== undefined) { + // The child only waits, so that the parent can reboot while it is running. + await setTimeout(childSleepSecs * 1000); +} else if (!childRunId) { + const run = await Actor.start(actorId, childInput, { runName: 'child' }); + await Actor.setValue('CHILD_RUN_ID', run.id); + + if (mode === 'resurrect') { + await Actor.apifyClient.run(run.id).abort(); + const abortedRun = await Actor.apifyClient.run(run.id).waitForFinish(); + assert.equal(abortedRun.status, 'ABORTED'); + } + + // The container gets killed while this waits, the run continues from the top. + await Actor.reboot(); +} else if (mode === 'resurrect') { + const run = await Actor.start(actorId, childInput, { runName: 'child' }); + try { + assert.equal(run.id, childRunId); + assert.ok(['READY', 'RUNNING'].includes(run.status), `run.status=${run.status}`); + } finally { + await Actor.apifyClient.run(run.id).abort(); + } + await Actor.setValue('VERIFIED', true); +} else { + assert.deepEqual(Object.keys(Actor.childRuns), ['child']); + assert.equal(Actor.childRuns.child.id, childRunId); + + const run = await Actor.call(actorId, childInput, { runName: 'child' }); + assert.equal(run.id, childRunId); + assert.equal(run.status, 'SUCCEEDED'); + await Actor.setValue('VERIFIED', true); +} + +await Actor.exit(); diff --git a/test/e2e/sdk/actorChildRuns/test.mjs b/test/e2e/sdk/actorChildRuns/test.mjs new file mode 100644 index 0000000000..af19796e40 --- /dev/null +++ b/test/e2e/sdk/actorChildRuns/test.mjs @@ -0,0 +1,37 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { ApifyClient } from 'apify'; + +const client = new ApifyClient({ + token: process.env.APIFY_TOKEN, +}); + +const actor = client.actor(process.argv[2]); + +const runParent = async (mode) => { + const { total: runsBefore } = await actor.runs().list(); + const run = await actor.call({ mode }); + const { total: runsAfter } = await actor.runs().list(); + const verified = await client.keyValueStore(run.defaultKeyValueStoreId).getRecord('VERIFIED'); + + return { run, startedRuns: runsAfter - runsBefore, verified: verified?.value === true }; +}; + +test('a named child run is reattached after a reboot', async () => { + const { run, startedRuns, verified } = await runParent('reattach'); + + assert.equal(run.status, 'SUCCEEDED'); + assert.ok(verified, 'the parent did not get past the reboot'); + // The parent run and the one child run it reattached to. + assert.equal(startedRuns, 2); +}); + +test('a named aborted child run is resurrected after a reboot', async () => { + const { run, startedRuns, verified } = await runParent('resurrect'); + + assert.equal(run.status, 'SUCCEEDED'); + assert.ok(verified, 'the parent did not get past the reboot'); + // The parent run and the one child run it resurrected. + assert.equal(startedRuns, 2); +});