diff --git a/src/resource_clients/log.ts b/src/resource_clients/log.ts index 81887f0fc..143af0026 100644 --- a/src/resource_clients/log.ts +++ b/src/resource_clients/log.ts @@ -13,7 +13,17 @@ import { ResourceClient } from '../base/resource_client.js'; 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 { cast, catchNotFoundForResourceOrThrow, concatBytes, parseArgument, sleep } from '../utils.js'; + +/** + * Pause before the first reopen of a log stream that ended before the run logged anything. Each further reopen doubles + * the pause up to `EMPTY_LOG_STREAM_MAX_RETRY_MILLIS`, which bounds the request rate while a run waits long to start, + * for example for free memory. + */ +const EMPTY_LOG_STREAM_RETRY_MILLIS = 500; + +/** Upper bound on the pause between reopens of an empty log stream. */ +const EMPTY_LOG_STREAM_MAX_RETRY_MILLIS = 5_000; const logOptionsSchema = z.strictObject({ raw: z.boolean().optional(), ...timeoutOptionsShape }); @@ -179,6 +189,7 @@ export class StreamedLog { #streamingTask: Promise | null = null; #stopLogging = false; #stopController = new AbortController(); + #wakeController = new AbortController(); constructor(options: StreamedLogOptions) { const { toLog, logClient, fromStart = true, signal } = options; @@ -197,17 +208,20 @@ export class StreamedLog { } this.#stopLogging = false; this.#stopController = new AbortController(); + this.#wakeController = new AbortController(); this.#streamingTask = this.#streamLog(); } /** - * Stop log redirection. Waits up to one second for lines already in flight, then aborts the log stream request. + * Stop log redirection. Waits up to one second for lines already in flight, then aborts the log stream request. If + * no stream has delivered anything yet, reads the whole log in one request instead. */ public async stop(): Promise { if (!this.#streamingTask) { throw new Error('Streaming task is not active'); } this.#stopLogging = true; + this.#wakeController.abort(); const stopController = this.#stopController; const abortTimeout = setTimeout(() => stopController.abort(), STOP_GRACE_MILLIS); try { @@ -230,11 +244,42 @@ export class StreamedLog { ? AbortSignal.any([this.#signal, this.#stopController.signal]) : this.#stopController.signal; try { - const logStream = await this.#logClient.stream({ raw: true, signal }); - if (!logStream) { - return; + let lastChunkRemainder: Uint8Array | undefined; + let retryMillis = EMPTY_LOG_STREAM_RETRY_MILLIS; + // The API serves the log of a run that has not logged anything yet as an empty stream that ends at once, + // so reopen it until the first bytes arrive. If stopped before any bytes arrive, fetch it in one request. + while (!lastChunkRemainder) { + if (this.#stopLogging) { + // `signal` is aborted once the stop grace period ends, which would cut this read short. + const logContent = await this.#logClient.get({ raw: true, signal: this.#signal }); + lastChunkRemainder = await this.#logStreamChunks( + [new TextEncoder().encode(logContent ?? '')], + signal, + ); + break; + } + let logStream: Readable | undefined; + try { + logStream = await this.#logClient.stream({ raw: true, signal }); + } catch (err) { + // The stop grace period ended while the stream was connecting, so read the log in one request. + if (this.#stopLogging && !this.#signal?.aborted) continue; + throw err; + } + if (!logStream) { + return; + } + // A stream opened during stop() would be cut after its first chunk, so read the log in one request. + if (this.#stopLogging) { + logStream.destroy(); + continue; + } + lastChunkRemainder = await this.#logStreamChunks(logStream, signal); + if (!lastChunkRemainder) { + await sleep(retryMillis, AbortSignal.any([signal, this.#wakeController.signal])); + retryMillis = Math.min(retryMillis * 2, EMPTY_LOG_STREAM_MAX_RETRY_MILLIS); + } } - const lastChunkRemainder = await this.#logStreamChunks(logStream, signal); // Process whatever is left when exiting. Maybe it is incomplete, maybe it is last log without EOL. const lastMessage = this.#decoder.decode(lastChunkRemainder).trim(); if (lastMessage.length) { @@ -246,14 +291,22 @@ export class StreamedLog { } } - async #logStreamChunks(logStream: Readable, signal: AbortSignal): Promise { + /** + * Redirect every complete message in the chunks and return the incomplete rest, or `undefined` when there were no + * chunks at all. + */ + async #logStreamChunks( + logStream: AsyncIterable | Iterable, + signal: AbortSignal, + ): Promise { // Chunk may be incomplete. Keep remainder for next chunk. - let previousChunkRemainder: Uint8Array = new Uint8Array(); + let previousChunkRemainder: Uint8Array | undefined; try { for await (const chunk of logStream) { // Handle possible leftover incomplete line from previous chunk. // Everything before last end of line is complete. + previousChunkRemainder ??= new Uint8Array(); const chunkWithPreviousRemainder = new Uint8Array(previousChunkRemainder.length + chunk.length); chunkWithPreviousRemainder.set(previousChunkRemainder, 0); chunkWithPreviousRemainder.set(chunk, previousChunkRemainder.length); diff --git a/test/actors.test.ts b/test/actors.test.ts index af619ef7f..cffd061ef 100644 --- a/test/actors.test.ts +++ b/test/actors.test.ts @@ -970,10 +970,20 @@ describe('Run actor with redirected logs', () => { let baseUrl: string; let client: ApifyClient; const statusGenerator = new StatusGenerator(); + // Streamed log requests left to answer with an empty body, the way the API answers before a run has logged anything. + let emptyLogStreams = 0; beforeAll(async () => { // Use custom router for the tests const router = express.Router(); + router.get('/actor-runs/redirect-run-id/log', (req, res, next) => { + if (req.query.stream && emptyLogStreams > 0) { + emptyLogStreams -= 1; + res.end(); + return; + } + next(); + }); // Set up a status generator to simulate run status changes. It will be reset for each test. router.get('/actor-runs/redirect-run-id', async (_, res) => { // Delay the response to give the actor time to run and produce expected logs @@ -1003,6 +1013,7 @@ describe('Run actor with redirected logs', () => { afterEach(async () => { // Reset the generator to so that the next test starts fresh statusGenerator.reset(); + emptyLogStreams = 0; client = null as unknown as ApifyClient; }); @@ -1035,5 +1046,35 @@ describe('Run actor with redirected logs', () => { expect(logSpy.mock.calls).toEqual([]); logSpy.mockRestore(); }); + + test('redirects the whole log when every log stream is empty until the run finishes', async () => { + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); + emptyLogStreams = Infinity; + + await client.actor('redirect-actor-id').call(undefined, { log: 'default' }); + + const prefix = c.cyan('redirect-actor-name runId:redirect-run-id -> '); + expect(logSpy.mock.calls).toEqual(MOCKED_ACTOR_LOGS_PROCESSED.map((item) => [prefix + item])); + logSpy.mockRestore(); + }); + }); + + describe('run.getStreamedLog - empty log stream', () => { + test('reopens the log stream when the first one is empty', async () => { + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); + emptyLogStreams = 1; + + const streamedLog = await client.run('redirect-run-id').getStreamedLog(); + streamedLog?.start(); + // Only the reopened stream can deliver the log before stop() reads the rest with a plain request. + await vi.waitFor(() => expect(logSpy).toHaveBeenCalledTimes(MOCKED_ACTOR_LOGS_PROCESSED.length), { + timeout: 10_000, + }); + await streamedLog?.stop(); + + const prefix = c.cyan('redirect-actor-name runId:redirect-run-id -> '); + expect(logSpy.mock.calls).toEqual(MOCKED_ACTOR_LOGS_PROCESSED.map((item) => [prefix + item])); + logSpy.mockRestore(); + }); }); }); diff --git a/test/logs.test.ts b/test/logs.test.ts index 3e6bf52e5..5165827f0 100644 --- a/test/logs.test.ts +++ b/test/logs.test.ts @@ -1,8 +1,14 @@ import type { AddressInfo } from 'node:net'; +import { Readable } from 'node:stream'; + +import log, { Log } from '@apify/log'; import { ApifyClient } from 'apify-client'; import type { Page } from 'puppeteer'; -import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test } from 'vitest'; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test, vi } from 'vitest'; + +import type { LogClient } from '../src/resource_clients/log.js'; +import { StreamedLog } from '../src/resource_clients/log.js'; import { DEFAULT_OPTIONS, asBrowserResult, Browser, validateRequest } from './_helper.js'; import { mockServer } from './mock_server/server.js'; @@ -87,3 +93,104 @@ describe('Log methods', () => { }); }); }); + +test('StreamedLog redirects the whole log when stop() lands while an empty log stream is being reopened', async () => { + const lines = [0, 1, 2].map((i) => `2025-01-01T00:00:0${i}.000Z line ${i}\n`); + const { promise: reopenedStreamGate, resolve: openReopenedStream } = Promise.withResolvers(); + const stream = vi + .fn() + .mockResolvedValueOnce(Readable.from([])) + .mockImplementationOnce(async () => { + await reopenedStreamGate; + return Readable.from(lines.map((line) => Buffer.from(line))); + }); + const logClient = { stream, get: vi.fn().mockResolvedValue(lines.join('')) } as unknown as LogClient; + const toLog = new Log(); + const info = vi.spyOn(toLog, 'info').mockImplementation(() => {}); + + const streamedLog = new StreamedLog({ toLog, logClient }); + streamedLog.start(); + await vi.waitFor(() => expect(stream).toHaveBeenCalledTimes(2), { timeout: 5_000 }); + const stopping = streamedLog.stop(); + openReopenedStream(); + await stopping; + + expect(info.mock.calls).toEqual(lines.map((line) => [line.trim()])); +}); + +test('StreamedLog doubles the pause between empty log streams up to its cap', async () => { + vi.useFakeTimers(); + try { + const streamTimes: number[] = []; + const stream = vi.fn().mockImplementation(async () => { + streamTimes.push(Date.now()); + const chunks = streamTimes.length <= 6 ? [] : [Buffer.from('2025-01-01T00:00:00.000Z line\n')]; + return Readable.from(chunks); + }); + const logClient = { stream, get: vi.fn() } as unknown as LogClient; + const toLog = new Log(); + vi.spyOn(toLog, 'info').mockImplementation(() => {}); + + const streamedLog = new StreamedLog({ toLog, logClient }); + streamedLog.start(); + await vi.advanceTimersByTimeAsync(17_500); + await streamedLog.stop(); + + const pauses = streamTimes.slice(1).map((time, i) => time - streamTimes[i]); + expect(pauses).toEqual([500, 1_000, 2_000, 4_000, 5_000, 5_000]); + } finally { + vi.useRealTimers(); + } +}); + +test('StreamedLog reads the whole log when the stop grace period ends while a log stream is connecting', async () => { + const lines = [0, 1, 2].map((i) => `2025-01-01T00:00:0${i}.000Z line ${i}\n`); + const stream = vi.fn().mockImplementation( + async ({ signal }: { signal: AbortSignal }) => + new Promise((_, reject) => { + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + }), + ); + const logClient = { stream, get: vi.fn().mockResolvedValue(lines.join('')) } as unknown as LogClient; + const toLog = new Log(); + const info = vi.spyOn(toLog, 'info').mockImplementation(() => {}); + + const streamedLog = new StreamedLog({ toLog, logClient }); + streamedLog.start(); + await vi.waitFor(() => expect(stream).toHaveBeenCalled()); + await streamedLog.stop(); + + expect(info.mock.calls).toEqual(lines.map((line) => [line.trim()])); +}); + +test('StreamedLog neither reopens nor reads a log that does not exist', async () => { + const stream = vi.fn().mockResolvedValue(undefined); + const get = vi.fn(); + const logClient = { stream, get } as unknown as LogClient; + + const streamedLog = new StreamedLog({ toLog: new Log(), logClient }); + streamedLog.start(); + await streamedLog.stop(); + + expect(stream).toHaveBeenCalledTimes(1); + expect(get).not.toHaveBeenCalled(); +}); + +test('StreamedLog reports a failing one-shot log read and still stops', async () => { + const stream = vi.fn().mockImplementation(async () => Readable.from([])); + const get = vi.fn().mockRejectedValue(new Error('Simulated log read failure')); + const logClient = { stream, get } as unknown as LogClient; + const warning = vi.spyOn(log, 'warning').mockImplementation(() => {}); + + try { + const streamedLog = new StreamedLog({ toLog: new Log(), logClient }); + streamedLog.start(); + await vi.waitFor(() => expect(stream).toHaveBeenCalled()); + await streamedLog.stop(); + + expect(get).toHaveBeenCalledTimes(1); + expect(warning).toHaveBeenCalledWith('Log redirection stopped due to error', expect.any(Error)); + } finally { + warning.mockRestore(); + } +});