diff --git a/docs/02_concepts/05_bundled-environments.md b/docs/02_concepts/05_bundled-environments.md index 44f810d01..d1ca673ea 100644 --- a/docs/02_concepts/05_bundled-environments.md +++ b/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 `LogClient.stream()` and `RunClient.getStreamedLog()`, and the `stream` option of `KeyValueStoreClient.getRecord()`, which all return a Node.js `Readable` stream. +- Status message redirection with `RunClient.getStatusMessageWatcher()`. 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. diff --git a/docs/public-api/apify-client.api.md b/docs/public-api/apify-client.api.md index c7c079600..7eb2987f0 100644 --- a/docs/public-api/apify-client.api.md +++ b/docs/public-api/apify-client.api.md @@ -2885,6 +2885,12 @@ type GeneratedWebhookDispatchEventData = NonNullable; get(options?: RunGetOptions): Promise; + getStatusMessageWatcher(options?: GetStatusMessageWatcherOptions): Promise; getStreamedLog(options?: GetStreamedLogOptions): Promise; 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; +} + +// @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); diff --git a/src/resource_clients/actor.ts b/src/resource_clients/actor.ts index 00a277348..8c0f878e4 100644 --- a/src/resource_clients/actor.ts +++ b/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(), + ]); + } } /** diff --git a/src/resource_clients/log.ts b/src/resource_clients/log.ts index 33694cf3b..710bd75ca 100644 --- a/src/resource_clients/log.ts +++ b/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 | 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 { + if (!this.#loggingTask) { + throw new Error('Logging task is not active'); + } + const { waitSecs = 0 } = options; + let graceTimer: ReturnType | undefined; + try { + if (waitSecs > 0) { + await Promise.race([ + this.#loggingTask, + new Promise((resolve) => { + graceTimer = setTimeout(resolve, waitSecs * 1000); + }), + ]); + } + this.#stopLogging = true; + this.#wakeUp?.(); + await this.#loggingTask; + } finally { + clearTimeout(graceTimer); + this.#loggingTask = null; + } + } + + async #logChangedStatusMessage(): Promise { + 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 { + if (this.#stopLogging) { + return; + } + await new Promise((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; +} diff --git a/src/resource_clients/run.ts b/src/resource_clients/run.ts index 6a4ed8b78..1afd65e54 100644 --- a/src/resource_clients/run.ts +++ b/src/resource_clients/run.ts @@ -6,7 +6,7 @@ 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'; @@ -14,7 +14,7 @@ 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 { + 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 { + 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. */ diff --git a/test/actors.test.ts b/test/actors.test.ts index f93f6d37b..2aa56141b 100644 --- a/test/actors.test.ts +++ b/test/actors.test.ts @@ -23,7 +23,7 @@ import { stringifyWebhooksToBase64 } from '../src/utils.js'; import { DEFAULT_OPTIONS, asBrowserResult, Browser, validateRequest } from './_helper.js'; import * as fixtures from './mock_server/fixtures.js'; import { createDefaultApp, mockServer } from './mock_server/server.js'; -import { MOCKED_ACTOR_LOGS_PROCESSED, StatusGenerator } from './mock_server/test_utils.js'; +import { MOCKED_ACTOR_LOGS_PROCESSED, MOCKED_ACTOR_STATUSES, StatusGenerator } from './mock_server/test_utils.js'; describe('Actor methods', () => { let baseUrl: string; @@ -942,9 +942,16 @@ describe('Run actor with redirected logs', () => { // Delay the response to give the actor time to run and produce expected logs await setTimeout(10); - const [status, statusMessage] = statusGenerator.next().value; + const [status, statusMessage, isStatusMessageTerminal] = statusGenerator.next().value; res.json({ - data: { ...fixtures.run, id: 'redirect-run-id', actId: 'redirect-actor-id', status, statusMessage }, + data: { + ...fixtures.run, + id: 'redirect-run-id', + actId: 'redirect-actor-id', + status, + statusMessage, + isStatusMessageTerminal, + }, }); }); const app = createDefaultApp(router); @@ -986,8 +993,34 @@ describe('Run actor with redirected logs', () => { await client.actor('redirect-actor-id').call(undefined, logOptions); - expect(logSpy.mock.calls).toEqual(MOCKED_ACTOR_LOGS_PROCESSED.map((item) => [expectedPrefix + item])); + const lines = logSpy.mock.calls.map(([line]) => line as string); + logSpy.mockRestore(); + const statusPrefix = `${expectedPrefix}Status: `; + const statusLines = lines.filter((line) => line.startsWith(statusPrefix)); + + expect(lines.filter((line) => !line.startsWith(statusPrefix))).toEqual( + MOCKED_ACTOR_LOGS_PROCESSED.map((item) => expectedPrefix + item), + ); + // Which statuses the watcher sees depends on polling timing, but it never repeats one and always + // ends with the terminal status message. + const possibleStatusLines = MOCKED_ACTOR_STATUSES.map( + ([status, statusMessage]) => `${statusPrefix}${status}, Message: ${statusMessage}`, + ); + expect(possibleStatusLines).toEqual(expect.arrayContaining(statusLines)); + expect(statusLines.every((line, i) => line !== statusLines[i - 1])).toBe(true); + expect(statusLines.at(-1)).toBe(`${statusPrefix}SUCCEEDED, Message: Actor Finished`); + }); + + test('call() skips the final status message wait when the run has not finished', async () => { + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); + + const callStarted = Date.now(); + const run = await client.actor('redirect-actor-id').call(undefined, { waitSecs: 0 }); + const callDuration = Date.now() - callStarted; logSpy.mockRestore(); + + expect(run.status).toBe('RUNNING'); + expect(callDuration).toBeLessThan(3000); }); test('logOptions:{ "log": null }', async () => { diff --git a/test/client_timeouts.test.ts b/test/client_timeouts.test.ts index ec5a82a2a..4641675c4 100644 --- a/test/client_timeouts.test.ts +++ b/test/client_timeouts.test.ts @@ -192,6 +192,13 @@ const TIMEOUT_CASES: TimeoutCase[] = [ // The method reads the run to find its Actor, and that lookup is what the tier times. responseData: { ...DEFAULT_RESPONSE_DATA, actId: 'actor-id' }, }, + { + client: 'RunClient', + method: 'getStatusMessageWatcher', + tier: 'long', + // The method reads the run to find its Actor, and that lookup is what the tier times. + responseData: { ...DEFAULT_RESPONSE_DATA, actId: 'actor-id' }, + }, { client: 'RunCollectionClient', method: 'list', tier: 'medium' }, { client: 'ScheduleClient', method: 'get', tier: 'short' }, { client: 'ScheduleClient', method: 'update', tier: 'short', args: [{ name: 'x' }] }, diff --git a/test/mock_server/test_utils.ts b/test/mock_server/test_utils.ts index ffef4a79d..6052e20d4 100644 --- a/test/mock_server/test_utils.ts +++ b/test/mock_server/test_utils.ts @@ -33,24 +33,24 @@ export const MOCKED_ACTOR_LOGS_PROCESSED = [ '2025-05-13T07:28:14.132Z [apify.redirect-logger runId:4U1oAnKau6jpzjUuA] -> 2025-05-13T07:27:14.132Z ACTOR:...', ]; -export const MOCKED_ACTOR_STATUSES = [ - ['RUNNING', 'Actor Started'], - ['RUNNING', 'Doing some stuff'], - ['RUNNING', 'Doing some stuff'], - ['RUNNING', 'Doing some stuff'], - ['RUNNING', 'Doing some stuff'], - ['RUNNING', 'Doing some stuff'], - ['RUNNING', 'Doing some stuff'], - ['RUNNING', 'Doing some stuff'], - ['RUNNING', 'Doing some stuff'], - ['SUCCEEDED', 'Actor Finished'], +export const MOCKED_ACTOR_STATUSES: [status: string, statusMessage: string, isStatusMessageTerminal: boolean][] = [ + ['RUNNING', 'Actor Started', false], + ['RUNNING', 'Doing some stuff', false], + ['RUNNING', 'Doing some stuff', false], + ['RUNNING', 'Doing some stuff', false], + ['RUNNING', 'Doing some stuff', false], + ['RUNNING', 'Doing some stuff', false], + ['RUNNING', 'Doing some stuff', false], + ['RUNNING', 'Doing some stuff', false], + ['RUNNING', 'Doing some stuff', false], + ['SUCCEEDED', 'Actor Finished', true], ]; /** * Helper class to allow iterating over defined list of statuses for each test case. */ export class StatusGenerator { - private generator: Generator; + private generator: Generator<(typeof MOCKED_ACTOR_STATUSES)[number], void, unknown>; private *getStatusGenerator() { for (const status of MOCKED_ACTOR_STATUSES) { @@ -70,7 +70,7 @@ export class StatusGenerator { this.generator = this.getStatusGenerator(); } - next(): IteratorResult { + next(): IteratorResult<(typeof MOCKED_ACTOR_STATUSES)[number]> { return this.generator.next(); } } diff --git a/test/runs.test.ts b/test/runs.test.ts index 7b0d2e544..e6d7e3191 100644 --- a/test/runs.test.ts +++ b/test/runs.test.ts @@ -2,14 +2,17 @@ import type { AddressInfo } from 'node:net'; import { setTimeout as setTimeoutNode } from 'node:timers/promises'; import c from 'ansi-colors'; -import { ApifyApiError, ApifyClient, ArgumentValidationError } from 'apify-client'; +import { ApifyApiError, ApifyClient, ArgumentValidationError, LoggerActorRedirect } from 'apify-client'; +import express from 'express'; import type { Page } from 'puppeteer'; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test, vi } from 'vitest'; +import { LEVELS, Log } from '@apify/log'; + import { DEFAULT_OPTIONS, asBrowserResult, Browser, validateRequest } from './_helper.js'; import * as fixtures from './mock_server/fixtures.js'; -import { mockServer } from './mock_server/server.js'; -import { MOCKED_ACTOR_LOGS_PROCESSED } from './mock_server/test_utils.js'; +import { createDefaultApp, mockServer } from './mock_server/server.js'; +import { MOCKED_ACTOR_LOGS_PROCESSED, StatusGenerator } from './mock_server/test_utils.js'; describe('Run methods', () => { let baseUrl: string; @@ -381,6 +384,14 @@ describe('Run methods', () => { expect(browserRes).toBeUndefined(); }); + test('getStatusMessageWatcher() returns undefined outside Node.js', async () => { + const browserRes = await page.evaluate( + (rId) => client.run(rId).getStatusMessageWatcher(), + 'redirect-run-id', + ); + expect(browserRes).toBeUndefined(); + }); + test.each(['dataset', 'keyValueStore', 'requestQueue', 'log'] as const)( '%s().get() throws on 404 status code', async (method) => { @@ -501,3 +512,95 @@ describe('Redirect run logs', () => { }); }); }); + +describe('Redirect run status messages', () => { + let baseUrl: string; + let client: ApifyClient; + const statusGenerator = new StatusGenerator(); + const prefix = 'watched -> '; + const toLog = new Log({ level: LEVELS.DEBUG, prefix, logger: new LoggerActorRedirect() }); + + beforeAll(async () => { + const router = express.Router(); + router.get('/actor-runs/status-run-id', async (_, res) => { + const [status, statusMessage, isStatusMessageTerminal] = statusGenerator.next().value; + res.json({ + data: { ...fixtures.run, id: 'status-run-id', status, statusMessage, isStatusMessageTerminal }, + }); + }); + router.get('/actor-runs/failing-run-id', async (_, res) => { + res.status(500).json({ error: { type: 'internal-error', message: 'Boom' } }); + }); + const server = await mockServer.start(undefined, createDefaultApp(router)); + baseUrl = `http://localhost:${(server.address() as AddressInfo).port}`; + }); + + afterAll(async () => { + await mockServer.close(); + }); + + beforeEach(() => { + client = new ApifyClient({ baseUrl, maxRetries: 0, ...DEFAULT_OPTIONS }); + }); + + afterEach(() => { + statusGenerator.reset(); + }); + + test('logs each status change once and stops at the terminal status message', async () => { + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); + + const watcher = await client.run('status-run-id').getStatusMessageWatcher({ toLog, checkPeriodSecs: 0.01 }); + watcher?.start(); + const stopStarted = Date.now(); + await watcher?.stop({ waitSecs: 10 }); + + expect(logSpy.mock.calls).toEqual([ + [`${c.cyan(prefix)}Status: RUNNING, Message: Actor Started`], + [`${c.cyan(prefix)}Status: RUNNING, Message: Doing some stuff`], + [`${c.cyan(prefix)}Status: SUCCEEDED, Message: Actor Finished`], + ]); + // The watcher ends by itself at the terminal status message, well before the grace period runs out. + expect(Date.now() - stopStarted).toBeLessThan(5000); + logSpy.mockRestore(); + }); + + test('stop() does not wait for the rest of the polling period', async () => { + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); + + const watcher = await client.run('status-run-id').getStatusMessageWatcher({ toLog, checkPeriodSecs: 60 }); + watcher?.start(); + await setTimeoutNode(100); + const stopStarted = Date.now(); + await watcher?.stop(); + + expect(Date.now() - stopStarted).toBeLessThan(1000); + expect(logSpy.mock.calls).toEqual([[`${c.cyan(prefix)}Status: RUNNING, Message: Actor Started`]]); + logSpy.mockRestore(); + }); + + test('logs a warning instead of throwing when polling fails', async () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + const watcher = await client.run('failing-run-id').getStatusMessageWatcher({ toLog }); + watcher?.start(); + await expect(watcher?.stop({ waitSecs: 5 })).resolves.toBeUndefined(); + + expect( + warnSpy.mock.calls.some( + ([msg]) => typeof msg === 'string' && msg.includes('Status message redirection stopped due to error'), + ), + ).toBe(true); + warnSpy.mockRestore(); + }); + + test('rejects a non-positive checkPeriodSecs', async () => { + await expect( + client.run('status-run-id').getStatusMessageWatcher({ toLog, checkPeriodSecs: 0 }), + ).rejects.toThrow(ArgumentValidationError); + }); + + test('returns undefined when toLog is null', async () => { + await expect(client.run('status-run-id').getStatusMessageWatcher({ toLog: null })).resolves.toBeUndefined(); + }); +});