fix: Turn any log streaming related errors to warnings (#876)

### Description
Errors in log streaming should not propagate further. Just log them as
warnings.
Such errors should be fixed on API level:
apify/apify-core#26653

### Issues
Closes: #873

Author: Pijukatel

src/resource_clients/log.ts

@@ -4,7 +4,7 @@ import type { Readable } from 'node:stream';
 import c from 'ansi-colors';
 
 import type { Log } from '@apify/log';
-import { Logger, LogLevel } from '@apify/log';
+import log, { Logger, LogLevel } from '@apify/log';
 
 import type { ApifyApiError } from '../apify_api_error';
 import type { ApiClientSubResourceOptions } from '../base/api_client';
@@ -197,11 +197,15 @@ export class StreamedLog {
         if (!logStream) {
             return;
         }
-        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 = Buffer.from(lastChunkRemainder).toString().trim();
-        if (lastMessage.length) {
-            this.destinationLog.info(lastMessage);
+        try {
+            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 = Buffer.from(lastChunkRemainder).toString().trim();
+            if (lastMessage.length) {
+                this.destinationLog.info(lastMessage);
+            }
+        } catch (err) {
+            log.warning(`Log redirection stopped due to error`, err as Error);
         }
     }
 

test/mock_server/server.ts

@@ -106,6 +106,18 @@ export function createDefaultApp(v2Router = express.Router()) {
         res.json({ data: { id: 'redirect-run-id', actId: 'redirect-actor-id', status: 'SUCCEEDED' } });
     });
 
+    v2Router.use('/actor-runs/econnreset-run-id/log', async (req: express.Request, res: express.Response) => {
+        res.write(MOCKED_ACTOR_LOGS[0]);
+        (res as any).flush();
+        await new Promise<void>((resolve) => {
+            setTimeout(resolve, 10);
+        });
+        req.socket.destroy();
+    });
+    v2Router.use('/actor-runs/econnreset-run-id', async (_, res) => {
+        res.json({ data: { id: 'econnreset-run-id', actId: 'redirect-actor-id', status: 'SUCCEEDED' } });
+    });
+
     v2Router.use('/actor-runs', runs);
     v2Router.use('/actor-tasks', tasks);
     v2Router.use('/users', users);

test/runs.test.ts

@@ -447,4 +447,18 @@ describe('Redirect run logs', () => {
             logSpy.mockRestore();
         });
     });
+
+    describe('run.getStreamedLog ECONNRESET', () => {
+        test('logs warning instead of throwing on error', async () => {
+            const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
+            const streamedLog = await client.run('econnreset-run-id').getStreamedLog({ fromStart: true });
+            streamedLog?.start();
+            await setTimeoutNode(500);
+            await expect(streamedLog?.stop()).resolves.not.toThrow();
+            expect(
+                warnSpy.mock.calls.some(([msg]: [string]) => msg?.includes('Log redirection stopped due to error')),
+            ).toBe(true);
+            warnSpy.mockRestore();
+        });
+    });
 });