feat: add RunClient.getStatusMessageWatcher()

docs/02_concepts/05_bundled-environments.md

@@ -34,6 +34,7 @@ Reaching the ES module build takes a bundler that doesn't set the `browser` cond
 These features rely on Node.js APIs and aren't available in the browser bundle or in the Web API runtime:
 
 - Log streaming with <ApiLink to="class/LogClient#stream">`LogClient.stream()`</ApiLink> and <ApiLink to="class/RunClient#getStreamedLog">`RunClient.getStreamedLog()`</ApiLink>, and the `stream` option of <ApiLink to="class/KeyValueStoreClient#getRecord">`KeyValueStoreClient.getRecord()`</ApiLink>, which all return a Node.js `Readable` stream.
+- Status message redirection with <ApiLink to="class/RunClient#getStatusMessageWatcher">`RunClient.getStatusMessageWatcher()`</ApiLink>. It pairs with `getStreamedLog()` in `ActorClient.call()`, so it returns `undefined` outside Node.js too.
 - Proxy support through the `HTTP_PROXY`, `HTTPS_PROXY`, and `NO_PROXY` environment variables.
 - Request body compression.
 - The `User-Agent` header, which browsers don't let a page set.

docs/public-api/apify-client.api.md

@@ -2885,6 +2885,12 @@ type GeneratedWebhookDispatchEventData = NonNullable<Schemas['WebhookDispatch'][
 // @public (undocumented)
 type GeneratedWebhookStats = Schemas['WebhookStats'];
 
+// @public
+export interface GetStatusMessageWatcherOptions extends TimeoutOptions {
+    checkPeriodSecs?: number;
+    toLog?: Log | null | 'default';
+}
+
 // @public
 export interface GetStreamedLogOptions extends TimeoutOptions {
     // (undocumented)
@@ -3656,6 +3662,7 @@ export class RunClient extends ResourceClient {
     dataset(): DatasetClient;
     delete(options?: TimeoutOptions): Promise<void>;
     get(options?: RunGetOptions): Promise<ActorRun | undefined>;
+    getStatusMessageWatcher(options?: GetStatusMessageWatcherOptions): Promise<StatusMessageWatcher | undefined>;
     getStreamedLog(options?: GetStreamedLogOptions): Promise<StreamedLog | undefined>;
     keyValueStore(): KeyValueStoreClient;
     log(): LogClient;
@@ -3850,6 +3857,25 @@ class Statistics {
     requests: number;
 }
 
+// @public
+export class StatusMessageWatcher {
+    constructor(options: StatusMessageWatcherOptions);
+    start(): void;
+    stop(options?: StatusMessageWatcherStopOptions): Promise<void>;
+}
+
+// @public
+export interface StatusMessageWatcherOptions {
+    checkPeriodSecs?: number;
+    runClient: RunClient;
+    toLog: Log;
+}
+
+// @public
+export interface StatusMessageWatcherStopOptions {
+    waitSecs?: number;
+}
+
 // @public
 export class StoreCollectionClient extends ResourceCollectionClient {
     constructor(options: ApiClientSubResourceOptions);

src/resource_clients/actor.ts

@@ -1,6 +1,6 @@
 import { z } from 'zod';
 
-import { ACTOR_JOB_STATUSES, ACTOR_PERMISSION_LEVEL, META_ORIGINS } from '@apify/consts';
+import { ACT_JOB_TERMINAL_STATUSES, ACTOR_JOB_STATUSES, ACTOR_PERMISSION_LEVEL, META_ORIGINS } from '@apify/consts';
 import { Log } from '@apify/log';
 
 import type { ApiClientSubResourceOptions } from '../base/api_client.js';
@@ -35,6 +35,10 @@ const startOptionsSchema = z.strictObject({
     forcePermissionLevel: z.enum(ACTOR_PERMISSION_LEVEL).optional(),
     ...timeoutOptionsShape,
 });
+
+// A run can set its final status message shortly after it finishes.
+const FINAL_STATUS_MESSAGE_WAIT_SECS = 6;
+
 const callOptionsSchema = z.strictObject({
     build: z.string().optional(),
     contentType: z.string().optional(),
@@ -263,7 +267,7 @@ export class ActorClient extends ResourceClient {
      * @param input - Input for the Actor, serialized to JSON. Omit it to run the Actor without input.
      * @param options - Run configuration options (extends all options from {@link start})
      * @param options.waitSecs - Maximum time to wait for the run to finish, in seconds. If omitted, waits indefinitely.
-     * @param options.log - Log instance for streaming run logs. Use `'default'` for console output, `null` to disable logging, or provide a custom Log instance.
+     * @param options.log - Log instance for streaming run logs and status messages. Use `'default'` for console output, `null` to disable logging, or provide a custom Log instance.
      * @param options.build - Tag or number of the build to run (e.g., `'beta'` or `'1.2.345'`).
      * @param options.memory - Memory in megabytes allocated for the run.
      * @param options.runTimeoutSecs - Maximum run duration in seconds.
@@ -301,14 +305,25 @@ export class ActorClient extends ResourceClient {
         // setting it up as a nested route under actor API.
         const newRunClient = this.apifyClient.run(id);
 
-        const streamedLog = await newRunClient.getStreamedLog({ toLog: log });
+        const [statusMessageWatcher, streamedLog] = await Promise.all([
+            newRunClient.getStatusMessageWatcher({ toLog: log }),
+            newRunClient.getStreamedLog({ toLog: log }),
+        ]);
+        statusMessageWatcher?.start();
         streamedLog?.start();
-        return this.apifyClient
-            .run(id)
-            .waitForFinish({ waitSecs, timeoutSecs })
-            .finally(async () => {
-                await streamedLog?.stop();
-            });
+        let finishedRun: ActorRun | undefined;
+        try {
+            finishedRun = await newRunClient.waitForFinish({ waitSecs, timeoutSecs });
+            return finishedRun;
+        } finally {
+            const runFinished =
+                finishedRun &&
+                ACT_JOB_TERMINAL_STATUSES.includes(finishedRun.status as (typeof ACT_JOB_TERMINAL_STATUSES)[number]);
+            await Promise.all([
+                statusMessageWatcher?.stop({ waitSecs: runFinished ? FINAL_STATUS_MESSAGE_WAIT_SECS : 0 }),
+                streamedLog?.stop(),
+            ]);
+        }
     }
 
     /**

src/resource_clients/log.ts

@@ -14,6 +14,7 @@ import type { ApifyRequestConfig } from '../http_clients/index.js';
 import type { TimeoutOptions } from '../timeouts.js';
 import { timeoutOptionsShape } from '../timeouts.js';
 import { cast, catchNotFoundForResourceOrThrow, concatBytes, parseArgument } from '../utils.js';
+import type { RunClient } from './run.js';
 
 const logOptionsSchema = z.strictObject({ raw: z.boolean().optional(), ...timeoutOptionsShape });
 
@@ -296,3 +297,130 @@ export interface StreamedLogOptions {
     /** Whether to redirect all logs from Actor run start (even logs from the past). */
     fromStart?: boolean;
 }
+
+/**
+ * Helper class for redirecting the status and status message of an Actor run to another log.
+ *
+ * The run is polled at a fixed interval and a line is logged whenever its status or status message changes, so
+ * a status message that is replaced within one polling period is never logged. Polling ends by itself once the run
+ * reports a terminal status message, or when {@link StatusMessageWatcher.stop} is called.
+ *
+ * Obtain an instance via {@link RunClient.getStatusMessageWatcher}.
+ */
+export class StatusMessageWatcher {
+    #destinationLog: Log;
+    #runClient: RunClient;
+    #checkPeriodSecs: number;
+    #lastStatusMessage = '';
+
+    #loggingTask: Promise<void> | null = null;
+    #stopLogging = false;
+    #wakeUp: (() => void) | null = null;
+
+    constructor(options: StatusMessageWatcherOptions) {
+        const { toLog, runClient, checkPeriodSecs = 1 } = options;
+        this.#destinationLog = toLog;
+        this.#runClient = runClient;
+        this.#checkPeriodSecs = checkPeriodSecs;
+    }
+
+    /**
+     * Start status message redirection.
+     */
+    public start(): void {
+        if (this.#loggingTask) {
+            throw new Error('Logging task already active');
+        }
+        this.#stopLogging = false;
+        this.#loggingTask = this.#logChangedStatusMessage();
+    }
+
+    /**
+     * Stop status message redirection.
+     *
+     * @param options.waitSecs - How long to keep polling for a terminal status message before stopping. A run can
+     * set its final status message shortly after it finishes, so waiting a few seconds makes it more likely to be
+     * logged. Default is `0`.
+     */
+    public async stop(options: StatusMessageWatcherStopOptions = {}): Promise<void> {
+        if (!this.#loggingTask) {
+            throw new Error('Logging task is not active');
+        }
+        const { waitSecs = 0 } = options;
+        let graceTimer: ReturnType<typeof setTimeout> | undefined;
+        try {
+            if (waitSecs > 0) {
+                await Promise.race([
+                    this.#loggingTask,
+                    new Promise<void>((resolve) => {
+                        graceTimer = setTimeout(resolve, waitSecs * 1000);
+                    }),
+                ]);
+            }
+            this.#stopLogging = true;
+            this.#wakeUp?.();
+            await this.#loggingTask;
+        } finally {
+            clearTimeout(graceTimer);
+            this.#loggingTask = null;
+        }
+    }
+
+    async #logChangedStatusMessage(): Promise<void> {
+        try {
+            while (!this.#stopLogging) {
+                const run = await this.#runClient.get();
+                if (run) {
+                    const message = `Status: ${run.status}, Message: ${run.statusMessage ?? ''}`;
+                    if (message !== this.#lastStatusMessage) {
+                        this.#lastStatusMessage = message;
+                        this.#destinationLog.info(message);
+                    }
+                    if (run.isStatusMessageTerminal) {
+                        return;
+                    }
+                }
+                await this.#sleep();
+            }
+        } catch (err) {
+            log.warning(`Status message redirection stopped due to error`, err as Error);
+        }
+    }
+
+    /**
+     * Wait one polling period, or less if `stop()` is called meanwhile.
+     */
+    async #sleep(): Promise<void> {
+        if (this.#stopLogging) {
+            return;
+        }
+        await new Promise<void>((resolve) => {
+            const timer = setTimeout(resolve, this.#checkPeriodSecs * 1000);
+            this.#wakeUp = () => {
+                clearTimeout(timer);
+                resolve();
+            };
+        });
+        this.#wakeUp = null;
+    }
+}
+
+/**
+ * Options for creating a {@link StatusMessageWatcher}.
+ */
+export interface StatusMessageWatcherOptions {
+    /** Run client used to poll the Actor run. */
+    runClient: RunClient;
+    /** Log to which the status and status message of the Actor run will be redirected. */
+    toLog: Log;
+    /** How often to poll the Actor run, in seconds. Default is `1`. */
+    checkPeriodSecs?: number;
+}
+
+/**
+ * Options for {@link StatusMessageWatcher.stop}.
+ */
+export interface StatusMessageWatcherStopOptions {
+    /** How long to keep polling for a terminal status message before stopping, in seconds. Default is `0`. */
+    waitSecs?: number;
+}

src/resource_clients/run.ts

@@ -6,15 +6,15 @@ import { LEVELS, Log } from '@apify/log';
 import type { ApiClientOptionsWithOptionalResourcePath } from '../base/api_client.js';
 import { ResourceClient } from '../base/resource_client.js';
 import type { ApifyRequestConfig, ApifyResponse } from '../http_clients/index.js';
-import type { TimeoutOptions } from '../timeouts.js';
+import type { Timeout, TimeoutOptions } from '../timeouts.js';
 import * as schemas from '../schemas.js';
 import { optionalTimeoutSchema, timeoutOptionsSchema, timeoutOptionsShape } from '../timeouts.js';
 import { runtime } from '#runtime';
 import { anyObjectSchema, parseArgument, parseResponse } from '../utils.js';
 import type { ActorInput, ActorRun } from './actor.js';
 import { DatasetClient } from './dataset.js';
 import { KeyValueStoreClient } from './key_value_store.js';
-import { LogClient, LoggerActorRedirect, StreamedLog } from './log.js';
+import { LogClient, LoggerActorRedirect, StatusMessageWatcher, StreamedLog } from './log.js';
 import { RequestQueueClient } from './request_queue.js';
 
 const RUN_CHARGE_IDEMPOTENCY_HEADER = 'idempotency-key';
@@ -494,22 +494,62 @@ export class RunClient extends ResourceClient {
             return undefined;
         }
         if (toLog === undefined || toLog === 'default') {
-            // Create default StreamedLog
-            // Get actor name and run id
-            const runData = await this.get({ timeoutSecs });
-            const runId = runData?.id ?? '';
-
-            const actorId = runData?.actId ?? '';
-            // `apifyClient.actor()` rejects an empty ID, which is what a run that could not be read leaves here.
-            const actorData = actorId ? await this.apifyClient.actor(actorId).get({ timeoutSecs }) : undefined;
-            const actorName = actorData?.name ?? '';
-            const name = [actorName, `runId:${runId}`].filter(Boolean).join(' ');
-
-            toLog = new Log({ level: LEVELS.DEBUG, prefix: `${name} -> `, logger: new LoggerActorRedirect() });
+            toLog = await this.#createRedirectLog(timeoutSecs);
         }
 
         return new StreamedLog({ logClient: this.log(), toLog, fromStart });
     }
+
+    /**
+     * Get StatusMessageWatcher for redirecting the run's status and status message to a log.
+     *
+     * @param options - Watcher options
+     * @param options.toLog - Log instance to redirect the status messages to. Use `'default'` for a preconfigured one, or `null` to disable the redirection.
+     * @param options.checkPeriodSecs - How often to poll the run, in seconds. Default is `1`.
+     * @param options.timeoutSecs - Timeout for the API requests that fetch the run and its Actor. Default is `'long'`.
+     *
+     * @example
+     * ```javascript
+     * const watcher = await client.run('run-id').getStatusMessageWatcher();
+     * watcher?.start();
+     * await client.run('run-id').waitForFinish();
+     * await watcher?.stop({ waitSecs: 6 });
+     * ```
+     */
+    async getStatusMessageWatcher(
+        options: GetStatusMessageWatcherOptions = {},
+    ): Promise<StatusMessageWatcher | undefined> {
+        parseArgument(options.timeoutSecs, optionalTimeoutSchema);
+        parseArgument(options.checkPeriodSecs, z.number().positive().optional());
+
+        const { checkPeriodSecs, timeoutSecs = 'long' } = options;
+        let { toLog } = options;
+        if (toLog === null || !runtime.isNode) {
+            // Explicitly no logging or not in Node.js
+            return undefined;
+        }
+        if (toLog === undefined || toLog === 'default') {
+            toLog = await this.#createRedirectLog(timeoutSecs);
+        }
+
+        return new StatusMessageWatcher({ runClient: this, toLog, checkPeriodSecs });
+    }
+
+    /**
+     * Create a log prefixed with the run's Actor name and run ID.
+     */
+    async #createRedirectLog(timeoutSecs: Timeout): Promise<Log> {
+        const runData = await this.get({ timeoutSecs });
+        const runId = runData?.id ?? '';
+
+        const actorId = runData?.actId ?? '';
+        // `apifyClient.actor()` rejects an empty ID, which is what a run that could not be read leaves here.
+        const actorData = actorId ? await this.apifyClient.actor(actorId).get({ timeoutSecs }) : undefined;
+        const actorName = actorData?.name ?? '';
+        const name = [actorName, `runId:${runId}`].filter(Boolean).join(' ');
+
+        return new Log({ level: LEVELS.DEBUG, prefix: `${name} -> `, logger: new LoggerActorRedirect() });
+    }
 }
 
 /**
@@ -521,6 +561,16 @@ export interface GetStreamedLogOptions extends TimeoutOptions {
     fromStart?: boolean;
 }
 
+/**
+ * Options for getting a status message watcher.
+ */
+export interface GetStatusMessageWatcherOptions extends TimeoutOptions {
+    /** Log to redirect the status messages to. Use `'default'` for a preconfigured one, or `null` to disable the redirection. */
+    toLog?: Log | null | 'default';
+    /** How often to poll the run, in seconds. Default is `1`. */
+    checkPeriodSecs?: number;
+}
+
 /**
  * Options for getting a Run.
  */