fix: redirect the run log when the first log stream comes back empty

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<void> | 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<void> {
         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<Uint8Array> {
+    /**
+     * 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<Uint8Array> | Iterable<Uint8Array>,
+        signal: AbortSignal,
+    ): Promise<Uint8Array | undefined> {
         // 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);

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();
+        });
     });
 });

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<void>();
+    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();
+    }
+});