feat: enhance performance and error handling in download engine with improved stream management and chunk processing

Author: ido-pluto

src/download/download-engine/download-file/download-engine-file.ts

@@ -1,12 +1,12 @@
 import { EventEmitter } from "../../../utils/EventEmitter.js";
 import BaseDownloadEngineWriteStream from "../streams/download-engine-write-stream/base-download-engine-write-stream.js";
-import {ChunkStatus, DownloadFile, SaveProgressInfo} from "../types.js";
+import { ChunkStatus, DownloadFile, SaveProgressInfo } from "../types.js";
 import BaseDownloadProgram from "./download-programs/base-download-program.js";
-import switchProgram, {AvailablePrograms} from "./download-programs/switch-program.js";
-import {DownloaderProgramManager} from "./downloaderProgramManager.js";
-import {DownloadFlags, DownloadStatus, ProgressStatus} from "./progress-status-file.js";
-import {pushComment} from "./utils/push-comment.js";
-import {randomUUID} from "crypto";
+import switchProgram, { AvailablePrograms } from "./download-programs/switch-program.js";
+import { DownloaderProgramManager } from "./downloaderProgramManager.js";
+import { DownloadFlags, DownloadStatus, ProgressStatus } from "./progress-status-file.js";
+import { pushComment } from "./utils/push-comment.js";
+import { randomUUID } from "crypto";
 
 export type DownloadEngineFilePerPartOptions = {
     parallelStreams: number;
@@ -88,7 +88,7 @@ export default class DownloadEngineFile extends EventEmitter<DownloadEngineFileE
     public constructor(file: DownloadFile, options: DownloadEngineFileOptions) {
         super();
         this.file = file;
-        this.options = {...DEFAULT_OPTIONS, ...options};
+        this.options = { ...DEFAULT_OPTIONS, ...options };
         this._progressStatus = {
             totalDownloadParts: file.parts.length,
             fileName: file.localFileName,
@@ -132,23 +132,35 @@ export default class DownloadEngineFile extends EventEmitter<DownloadEngineFileE
     }
 
     public get status(): ProgressStatus {
+        let retrying = false;
+        let retryingTotalAttempts = 0;
+        let streamsNotResponding = 0;
+        let streamingBytes = 0;
+
+        for (const key in this._activeStreamContext) {
+            const streamContext = this._activeStreamContext[key];
+            streamingBytes += streamContext.streamBytes;
+            retrying ||= Boolean(streamContext.isRetrying);
+            retryingTotalAttempts = Math.max(retryingTotalAttempts, streamContext.retryingAttempts);
+            streamsNotResponding += Number(Boolean(streamContext.isStreamNotResponding));
+        }
+
+        const transferredBytes = this._getAllTransferredBytes(streamingBytes);
+
         const thisStatus: ProgressStatus = {
             ...this._progressStatus as ProgressStatus,
             transferAction: this._activePart.fetchStream.transferAction,
             downloadId: this._progress.downloadId,
             downloadPart: this._progress.part + 1,
-            transferredBytes: this.transferredBytes,
+            transferredBytes,
             totalBytes: this.downloadSize,
             downloadStatus: this._downloadStatus,
-            comment: this.options.comment
+            comment: this.options.comment,
+            retrying,
+            retryingTotalAttempts,
+            streamsNotResponding
         };
 
-        const streamContexts = Object.values(this._activeStreamContext);
-
-        thisStatus.retrying = streamContexts.some(c => c.isRetrying);
-        thisStatus.retryingTotalAttempts = Math.max(0, ...streamContexts.map(x => x.retryingAttempts));
-        thisStatus.streamsNotResponding = streamContexts.reduce((acc, cur) => acc + (cur.isStreamNotResponding ? 1 : 0), 0);
-
         return thisStatus;
     }
 
@@ -173,6 +185,10 @@ export default class DownloadEngineFile extends EventEmitter<DownloadEngineFileE
         const streamingBytes = Object.values(this._activeStreamContext)
             .reduce((acc, cur) => acc + cur.streamBytes, 0);
 
+        return this._getAllTransferredBytes(streamingBytes);
+    }
+
+    private _getAllTransferredBytes(streamingBytes: number) {
         const streamBytes = this._activeDownloadedChunkSize + streamingBytes;
         const streamBytesMin = Math.min(streamBytes, this._activePart.downloadSize || streamBytes);
 
@@ -304,7 +320,7 @@ export default class DownloadEngineFile extends EventEmitter<DownloadEngineFileE
     }
 
     protected async _downloadSlice(startChunk: number, endChunk: number) {
-        const getContext = () => this._activeStreamContext[startChunk] ??= {streamBytes: 0, retryingAttempts: 0};
+        const getContext = () => this._activeStreamContext[startChunk] ??= { streamBytes: 0, retryingAttempts: 0 };
 
         const fetchState = this._activePart.fetchStream.withSubState({
             chunkSize: this._progress.chunkSize,

src/download/download-engine/streams/download-engine-fetch-stream/base-download-engine-fetch-stream.ts

@@ -1,15 +1,14 @@
 import retry from "async-retry";
 import { EventEmitter } from "../../../../utils/EventEmitter.js";
-import {withLock} from "lifecycle-utils";
+import { withLock } from "lifecycle-utils";
 import prettyMillisecondsCompact from "../../../transfer-visualize/utils/prettyMSFast.js";
-import {AvailablePrograms} from "../../download-file/download-programs/switch-program.js";
-import {InputRange} from "../../engine/base-download-engine.js";
-import {sleepPromise} from "../../utils/sleepPromise.js";
+import { AvailablePrograms } from "../../download-file/download-programs/switch-program.js";
+import { InputRange } from "../../engine/base-download-engine.js";
+import { sleepPromise } from "../../utils/sleepPromise.js";
 import HttpError from "./errors/http-error.js";
 import StatusCodeError from "./errors/status-code-error.js";
-import {retryAsyncStatementSimple} from "./utils/retry-async-statement.js";
+import { retryAsyncStatementSimple } from "./utils/retry-async-statement.js";
 
-export const STREAM_NOT_RESPONDING_TIMEOUT = 1000 * 3;
 export const MIN_LENGTH_FOR_MORE_INFO_REQUEST = 1024 * 1024 * 3; // 3MB
 
 const TOKEN_EXPIRED_ERROR_CODES = [401, 403, 419, 440, 498, 499];
@@ -19,7 +18,15 @@ export type BaseDownloadEngineFetchStreamOptions = {
     retryFetchDownloadInfo?: retry.Options;
     range?: InputRange;
     /**
-     * Max wait for next data stream
+     * Interval read data and check if stream is not responding (default: 1s)
+     */
+    streamCheckInterval?: number;
+    /**
+     * Max wait for stream to respond with data before raising stream not responding event (default: 3s)
+     */
+    streamWaitAlert?: number;
+    /**
+     * Max wait for next data stream before aborting and retrying (default: 15s)
      */
     maxStreamWait?: number;
     /**
@@ -41,19 +48,19 @@ export type BaseDownloadEngineFetchStreamOptions = {
      */
     progressThrottleMs?: number;
 } & (
-    {
-        defaultFetchDownloadInfo?: { length: number, acceptRange: boolean; };
-    } |
-    {
-        /**
-         * Try different headers to see if any authentication is needed
-         */
-        tryHeaders?: Record<string, string>[];
-        /**
-         * Delay between trying different headers
-         */
-        tryHeadersDelay?: number;
-    });
+        {
+            defaultFetchDownloadInfo?: { length: number, acceptRange: boolean; };
+        } |
+        {
+            /**
+             * Try different headers to see if any authentication is needed
+             */
+            tryHeaders?: Record<string, string>[];
+            /**
+             * Delay between trying different headers
+             */
+            tryHeadersDelay?: number;
+        });
 
 export type DownloadInfoResponse = {
     length: number,
@@ -93,6 +100,8 @@ export type WriteCallback = (data: Uint8Array[], position: number, index: number
 
 const DEFAULT_OPTIONS: BaseDownloadEngineFetchStreamOptions = {
     retryOnServerError: true,
+    streamCheckInterval: 10,
+    streamWaitAlert: 1000 * 3,
     maxStreamWait: 1000 * 15,
     headersTimeout: 1000 * 30,
     retry: {
@@ -111,8 +120,7 @@ const DEFAULT_OPTIONS: BaseDownloadEngineFetchStreamOptions = {
     range: {
         start: 0,
         end: -1
-    },
-    progressThrottleMs: 15
+    }
 };
 
 export default abstract class BaseDownloadEngineFetchStream extends EventEmitter<BaseDownloadEngineFetchStreamEvents> {
@@ -127,13 +135,16 @@ export default abstract class BaseDownloadEngineFetchStream extends EventEmitter
     public aborted = false;
     protected _pausedResolve?: () => void;
     protected _cleanupClonedStateListeners?: () => void;
-    public errorCount = {value: 0};
+    public errorCount = { value: 0 };
     public lastFetchTime = 0;
     private _closed = false;
+    private _watchDogCalls = new Set<() => void>();
+    private _watchDogInterval?: NodeJS.Timeout;
 
     constructor(options: Partial<BaseDownloadEngineFetchStreamOptions> = {}) {
         super();
-        this.options = {...DEFAULT_OPTIONS, ...options};
+        this.options = { ...DEFAULT_OPTIONS, ...options };
+        this.watchDog = this.watchDog.bind(this);
         this.initEvents();
     }
 
@@ -189,6 +200,8 @@ export default abstract class BaseDownloadEngineFetchStream extends EventEmitter
             fetchStream._cleanupClonedStateListeners = undefined;
         };
 
+        this.watchDog = fetchStream.watchDog;
+
         return fetchStream;
     }
 
@@ -338,6 +351,25 @@ export default abstract class BaseDownloadEngineFetchStream extends EventEmitter
         return parsed.href;
     }
 
+    protected watchDog(callback: () => void) {
+        this._watchDogCalls.add(callback);
+        if (!this._watchDogInterval) {            
+            this._watchDogInterval = setInterval(() => {
+                for (const cb of this._watchDogCalls) {
+                    cb();
+                }
+            }, this.options.streamCheckInterval!);
+        }
+
+        return () => {
+            this._watchDogCalls.delete(callback);
+            if (this._watchDogCalls.size === 0 && this._watchDogInterval) {
+                clearInterval(this._watchDogInterval);
+                this._watchDogInterval = undefined;
+            }
+        };
+    }
+
     protected retryOnServerError(error: Error): error is StatusCodeError {
         return Boolean(this.options.retryOnServerError) && error instanceof StatusCodeError &&
             (error.statusCode >= 500 || error.statusCode === 429);