From 06bf2bb88870ab0e008ea164e0c7a0f52a77d88b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 2 Oct 2026 08:40:49 +0200 Subject: [PATCH 1/6] fix(log): redirect the run log when the first log stream comes back empty --- src/resource_clients/log.ts | 41 ++++++++++++++++++++++++++++++------- test/actors.test.ts | 41 +++++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 7 deletions(-) diff --git a/src/resource_clients/log.ts b/src/resource_clients/log.ts index 0b41734b0..3daf4e6c8 100644 --- a/src/resource_clients/log.ts +++ b/src/resource_clients/log.ts @@ -13,7 +13,10 @@ 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 reopening a log stream that ended before the run logged anything. */ +const EMPTY_LOG_STREAM_RETRY_MILLIS = 500; const logOptionsSchema = z.strictObject({ raw: z.boolean().optional(), ...timeoutOptionsShape }); @@ -175,6 +178,7 @@ export class StreamedLog { #signal: AbortSignal | undefined; #streamingTask: Promise | null = null; #stopLogging = false; + #stopController = new AbortController(); constructor(options: StreamedLogOptions) { const { toLog, logClient, fromStart = true, signal } = options; @@ -192,6 +196,7 @@ export class StreamedLog { throw new Error('Streaming task already active'); } this.#stopLogging = false; + this.#stopController = new AbortController(); this.#streamingTask = this.#streamLog(); } @@ -203,6 +208,7 @@ export class StreamedLog { throw new Error('Streaming task is not active'); } this.#stopLogging = true; + this.#stopController.abort(); try { await this.#streamingTask; } catch (err) { @@ -219,11 +225,25 @@ export class StreamedLog { */ async #streamLog(): Promise { try { - const logStream = await this.#logClient.stream({ raw: true, signal: this.#signal }); - if (!logStream) { - return; + let lastChunkRemainder: Uint8Array | undefined; + // 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. Once stopped, read whatever the log holds in one request. + while (!lastChunkRemainder) { + if (this.#stopLogging) { + const logContent = await this.#logClient.get({ raw: true, signal: this.#signal }); + lastChunkRemainder = await this.#logStreamChunks([new TextEncoder().encode(logContent ?? '')]); + break; + } + const logStream = await this.#logClient.stream({ raw: true, signal: this.#signal }); + if (!logStream) { + return; + } + lastChunkRemainder = await this.#logStreamChunks(logStream); + if (!lastChunkRemainder) { + const wakeSignals = [this.#stopController.signal, ...(this.#signal ? [this.#signal] : [])]; + await sleep(EMPTY_LOG_STREAM_RETRY_MILLIS, AbortSignal.any(wakeSignals)); + } } - const lastChunkRemainder = await this.#logStreamChunks(logStream); // 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) { @@ -235,13 +255,20 @@ export class StreamedLog { } } - async #logStreamChunks(logStream: Readable): 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, + ): Promise { // Chunk may be incomplete. Keep remainder for next chunk. - let previousChunkRemainder: Uint8Array = new Uint8Array(); + let previousChunkRemainder: Uint8Array | undefined; 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 7b251808a..63e8212f8 100644 --- a/test/actors.test.ts +++ b/test/actors.test.ts @@ -947,10 +947,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 @@ -980,6 +990,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; }); @@ -1012,5 +1023,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(); + }); }); }); From fe46ba786a7c3453cd1a110c936b7d093acc622d Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 2 Oct 2026 09:08:45 +0200 Subject: [PATCH 2/6] fix(log): keep the whole log when stop() lands during a stream reopen --- src/resource_clients/log.ts | 5 +++++ test/logs.test.ts | 32 +++++++++++++++++++++++++++++++- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/src/resource_clients/log.ts b/src/resource_clients/log.ts index 3daf4e6c8..78a83f92f 100644 --- a/src/resource_clients/log.ts +++ b/src/resource_clients/log.ts @@ -238,6 +238,11 @@ export class StreamedLog { 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); if (!lastChunkRemainder) { const wakeSignals = [this.#stopController.signal, ...(this.#signal ? [this.#signal] : [])]; diff --git a/test/logs.test.ts b/test/logs.test.ts index 3e6bf52e5..4820779ce 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 } 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,27 @@ 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()])); +}); From 89fe6240f801515d2e59a044cc023162bcc67af7 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 2 Oct 2026 13:53:23 +0200 Subject: [PATCH 3/6] test(log): cover a missing log being neither reopened nor read --- test/logs.test.ts | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/test/logs.test.ts b/test/logs.test.ts index 4820779ce..43621824b 100644 --- a/test/logs.test.ts +++ b/test/logs.test.ts @@ -117,3 +117,16 @@ test('StreamedLog redirects the whole log when stop() lands while an empty log s 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(); +}); From f38452495aec989f19b1910886f92de5ebe3d961 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 2 Oct 2026 13:53:23 +0200 Subject: [PATCH 4/6] test(log): cover a failing one-shot log read on stop --- test/logs.test.ts | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/test/logs.test.ts b/test/logs.test.ts index 43621824b..e13dc517b 100644 --- a/test/logs.test.ts +++ b/test/logs.test.ts @@ -1,7 +1,7 @@ import type { AddressInfo } from 'node:net'; import { Readable } from 'node:stream'; -import { Log } from '@apify/log'; +import log, { Log } from '@apify/log'; import { ApifyClient } from 'apify-client'; import type { Page } from 'puppeteer'; @@ -130,3 +130,22 @@ test('StreamedLog neither reopens nor reads a log that does not exist', async () 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(); + } +}); From 53a0bbf18e28255d1df0c4e3c7d09b9a1463ac54 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:46:13 +0200 Subject: [PATCH 5/6] fix(log): back off between empty log streams --- src/resource_clients/log.ts | 13 +++++++++++-- test/logs.test.ts | 25 +++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/src/resource_clients/log.ts b/src/resource_clients/log.ts index 78a83f92f..b8eb030f3 100644 --- a/src/resource_clients/log.ts +++ b/src/resource_clients/log.ts @@ -15,9 +15,16 @@ import type { TimeoutOptions } from '../timeouts.js'; import { timeoutOptionsShape } from '../timeouts.js'; import { cast, catchNotFoundForResourceOrThrow, concatBytes, parseArgument, sleep } from '../utils.js'; -/** Pause before reopening a log stream that ended before the run logged anything. */ +/** + * 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 }); /** @@ -226,6 +233,7 @@ export class StreamedLog { async #streamLog(): Promise { try { 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. Once stopped, read whatever the log holds in one request. while (!lastChunkRemainder) { @@ -246,7 +254,8 @@ export class StreamedLog { lastChunkRemainder = await this.#logStreamChunks(logStream); if (!lastChunkRemainder) { const wakeSignals = [this.#stopController.signal, ...(this.#signal ? [this.#signal] : [])]; - await sleep(EMPTY_LOG_STREAM_RETRY_MILLIS, AbortSignal.any(wakeSignals)); + await sleep(retryMillis, AbortSignal.any(wakeSignals)); + retryMillis = Math.min(retryMillis * 2, EMPTY_LOG_STREAM_MAX_RETRY_MILLIS); } } // Process whatever is left when exiting. Maybe it is incomplete, maybe it is last log without EOL. diff --git a/test/logs.test.ts b/test/logs.test.ts index e13dc517b..9d3022ab6 100644 --- a/test/logs.test.ts +++ b/test/logs.test.ts @@ -118,6 +118,31 @@ test('StreamedLog redirects the whole log when stop() lands while an empty log s 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 neither reopens nor reads a log that does not exist', async () => { const stream = vi.fn().mockResolvedValue(undefined); const get = vi.fn(); From 5c8164ee923e6fad53f5d0d75a0c8492b034c15c Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 9 Oct 2026 11:42:58 +0200 Subject: [PATCH 6/6] docs: clarify when StreamedLog reads the log in one request --- src/resource_clients/log.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/resource_clients/log.ts b/src/resource_clients/log.ts index f86955ca4..143af0026 100644 --- a/src/resource_clients/log.ts +++ b/src/resource_clients/log.ts @@ -247,7 +247,7 @@ export class StreamedLog { 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. Once stopped, read whatever the log holds in one request. + // 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.