fix: Fix file download hang (#3411)

- split from #3325 after review
(#3325 (comment))

Author: janbuchar

packages/http-crawler/src/internals/file-download.ts

@@ -1,15 +1,17 @@
 import { Transform } from 'node:stream';
-import { finished } from 'node:stream/promises';
 
 import type { BasicCrawlerOptions } from '@crawlee/basic';
 import { BasicCrawler, ContextPipeline } from '@crawlee/basic';
 import type { CrawlingContext, LoadedRequest, Request } from '@crawlee/core';
+import { ResponseWithUrl } from '@crawlee/http-client';
 import type { Dictionary } from '@crawlee/types';
 
 import type { ErrorHandler, GetUserDataFromRequest, InternalHttpHook, RequestHandler, RouterRoutes } from '../index.js';
 import { Router } from '../index.js';
 import { parseContentTypeFromResponse } from './utils.js';
 
+const kBodyDrained = Symbol('bodyDrained');
+
 export type FileDownloadErrorHandler<
     UserData extends Dictionary = any, // with default to Dictionary we cant use a typed router in untyped crawler
 > = ErrorHandler<FileDownloadCrawlingContext<UserData>>;
@@ -164,7 +166,13 @@ export class FileDownload extends BasicCrawler<FileDownloadCrawlingContext> {
                 ContextPipeline.create<CrawlingContext>().compose({
                     action: async (context) => this.initiateDownload(context),
                     cleanup: async (context) => {
-                        await (context.response.body ? finished(context.response.body as any) : Promise.resolve());
+                        if (!context.response.bodyUsed) {
+                            // Nobody consumed the body — cancel it so the
+                            // underlying connection can be released.
+                            await context.response.body?.cancel();
+                        }
+
+                        await (context as { [kBodyDrained]: Promise<void> })[kBodyDrained];
                     },
                 }),
         });
@@ -179,16 +187,43 @@ export class FileDownload extends BasicCrawler<FileDownloadCrawlingContext> {
 
         context.request.url = response.url;
 
+        const { response: trackedResponse, bodyDrained } = trackBodyConsumption(response);
+
         const contextExtension = {
             request: context.request as LoadedRequest<Request>,
-            response,
+            response: trackedResponse,
             contentType: { type, encoding },
+            [kBodyDrained]: bodyDrained,
         };
 
         return contextExtension;
     }
 }
 
+/**
+ * Wraps a Response so that we can track when the body stream has been fully
+ * consumed (or errored). Pipes the original body through a TransformStream;
+ * the readable side becomes the new Response body, and `pipeTo` gives us a
+ * promise that resolves once the body is fully read or cancelled.
+ */
+function trackBodyConsumption(response: Response): { response: ResponseWithUrl; bodyDrained: Promise<void> } {
+    if (!response.body) {
+        return { response, bodyDrained: Promise.resolve() };
+    }
+
+    const passthrough = new TransformStream();
+    const bodyDrained = response.body.pipeTo(passthrough.writable).catch(() => {});
+
+    const trackedResponse = new ResponseWithUrl(passthrough.readable, {
+        headers: response.headers,
+        status: response.status,
+        statusText: response.statusText,
+        url: response.url,
+    });
+
+    return { response: trackedResponse, bodyDrained };
+}
+
 /**
  * Creates new {@apilink Router} instance that works based on request labels.
  * This instance can then serve as a `requestHandler` of your {@apilink FileDownload}.

test/core/crawlers/file_download.test.ts

@@ -6,6 +6,7 @@ import { ReadableStream } from 'node:stream/web';
 import { setTimeout } from 'node:timers/promises';
 
 import { FileDownload } from '@crawlee/http';
+import { FetchHttpClient } from '@crawlee/http-client';
 import express from 'express';
 import { startExpressAppPromise } from 'test/shared/_helper.js';
 import { afterAll, beforeAll, expect, test } from 'vitest';
@@ -95,8 +96,9 @@ test('requestHandler - reading bytes synchronously', async () => {
 
     const fileUrl = new URL('/file?size=1024&seed=123', url).toString();
 
-    await crawler.run([fileUrl]);
+    const stats = await crawler.run([fileUrl]);
 
+    expect(stats.requestsFailed).toBe(0);
     expect(results).toHaveLength(1);
     expect(results[0].length).toBe(1024);
     expect(results[0]).toEqual(await ReadableStreamGenerator.getUint8Array(1024, 123));
@@ -116,25 +118,29 @@ test('requestHandler - streaming response body', async () => {
 
     const fileUrl = new URL('/file?size=1024&seed=456', url).toString();
 
-    await crawler.run([fileUrl]);
+    const stats = await crawler.run([fileUrl]);
 
+    expect(stats.requestsFailed).toBe(0);
     expect(result.length).toBe(1024);
     expect(result).toEqual(await ReadableStreamGenerator.getUint8Array(1024, 456));
 });
 
 test('requestHandler receives response', async () => {
+    const fileUrl = new URL('/file?size=1024&seed=321', url).toString();
+
     const crawler = new FileDownload({
         maxRequestRetries: 0,
         requestHandler: async ({ response }) => {
             expect(response?.headers.get('content-type')).toBe('application/octet-stream');
             expect(response?.status).toBe(200);
             expect(response?.statusText).toBe('OK');
+            expect(response?.url).toBe(fileUrl);
         },
     });
 
-    const fileUrl = new URL('/file?size=1024&seed=456', url).toString();
+    const stats = await crawler.run([fileUrl]);
 
-    await crawler.run([fileUrl]);
+    expect(stats.requestsFailed).toBe(0);
 });
 
 test('crawler waits for the stream to be consumed', async () => {
@@ -146,8 +152,12 @@ test('crawler waits for the stream to be consumed', async () => {
         },
     });
 
+    // Use FetchHttpClient so response.body is a real streaming ReadableStream
+    // (the default GotScrapingHttpClient buffers the entire response, making
+    // the body complete instantly and the test a no-op).
     const crawler = new FileDownload({
         maxRequestRetries: 0,
+        httpClient: new FetchHttpClient(),
         requestHandler: async ({ response }) => {
             pipelineWithCallbacks(response.body ?? ReadableStream.from([]), bufferingStream, (err) => {
                 if (!err) {
@@ -162,7 +172,9 @@ test('crawler waits for the stream to be consumed', async () => {
 
     // waits for a second after every kilobyte sent.
     const fileUrl = new URL(`/file?size=${5 * 1024}&seed=789&throttle=1000`, url).toString();
-    await crawler.run([fileUrl]);
+    const stats = await crawler.run([fileUrl]);
+
+    expect(stats.requestsFailed).toBe(0);
 
     // Wait for the stream to finish (pipeline is async)
     await new Promise<void>((resolve) => {