fix(impit-client): pause fromWeb stream to prevent early consumption (#3347)

Attaching a 'data' listener on Readable.fromWeb(response.body) switches
the stream into flowing mode. This can start consuming the response body
before Crawlee attaches its pipeline consumer, so initial chunks are
lost and the downstream sees a truncated stream. That also aswers why
only retries worked.

Fix: call responseStream.pause() right after registering the 'data'
handler. The stream stays paused until the pipeline/pipe attaches and
resumes it, preventing early consumption while still allowing progress
tracking.

Closes [WCC issue
#555](apify/store-website-content-crawler#555)

Author: nikitachapovskii-dev

packages/impit-client/src/index.ts

@@ -1,4 +1,4 @@
-import { Readable } from 'node:stream';
+import { pipeline, Readable, Transform } from 'node:stream';
 import { type ReadableStream } from 'node:stream/web';
 import { isGeneratorObject } from 'node:util/types';
 
@@ -210,19 +210,24 @@ export class ImpitHttpClient implements BaseHttpClient {
         const responseStream = Readable.fromWeb(response.body as ReadableStream<any>);
         let transferred = 0;
         const total = Number(response.headers.get('content-length') ?? 0);
-        responseStream.on('data', (chunk) => {
-            transferred += chunk.length;
+        const counter = new Transform({
+            transform(chunk, _enc, cb) {
+                transferred += chunk.length;
+                cb(null, chunk);
+            },
         });
 
-        const getDownloadProgress = () => {
-            return {
-                percent: Math.round((transferred / total) * 100),
-                transferred,
-                total,
-            };
-        };
+        pipeline(responseStream, counter, (err) => {
+            if (err) counter.destroy(err);
+        });
+
+        const getDownloadProgress = () => ({
+            percent: total > 0 ? Math.round((transferred / total) * 100) : 0,
+            transferred,
+            total,
+        });
 
-        return [responseStream, getDownloadProgress];
+        return [counter, getDownloadProgress];
     }
 
     /**