fixup! Add stall detection to recover from frozen uploads

Author: m7kvqbe1

lib/StallDetector.ts

@@ -10,7 +10,6 @@ export class StallDetector {
   private intervalId: ReturnType<typeof setInterval> | null = null
   private lastProgressTime = 0
   private isActive = false
-  private isPaused = false
 
   constructor(
     options: StallDetectionOptions,
@@ -47,7 +46,7 @@ export class StallDetector {
 
     // Setup periodic check
     this.intervalId = setInterval(() => {
-      if (!this.isActive || this.isPaused) {
+      if (!this.isActive) {
         return
       }
 
@@ -76,26 +75,6 @@ export class StallDetector {
     this.lastProgressTime = Date.now()
   }
 
-  /**
-   * Pause stall detection temporarily (e.g., during onBeforeRequest callback)
-   */
-  pause(): void {
-    this.isActive = false
-    this.isPaused = true
-  }
-
-  /**
-   * Resume stall detection after pause
-   */
-  resume(): void {
-    if (this.isPaused) {
-      this.isPaused = false
-      this.isActive = true
-      // Reset the last progress time to avoid false positives
-      this.lastProgressTime = Date.now()
-    }
-  }
-
   /**
    * Detect if current HttpStack supports progress events
    */

lib/options.ts

@@ -153,8 +153,8 @@ export interface HttpStack {
   createRequest(method: string, url: string): HttpRequest
   getName(): string
 
-  // Indicates whether this HTTP stack implementation supports progress events
-  // during upload. If false, stall detection will use overall transfer rate instead.
+  // Indicates whether this HTTP stack implementation
+  // supports progress events during upload.
   supportsProgressEvents: () => boolean
 }
 

lib/upload.ts

@@ -19,10 +19,6 @@ import {
 } from './options.js'
 import { uuid } from './uuid.js'
 
-interface ExtendedHttpRequest extends HttpRequest {
-  _upload?: BaseUpload
-}
-
 export const defaultOptions = {
   endpoint: undefined,
 
@@ -120,9 +116,6 @@ export class BaseUpload {
   // upload options or HEAD response)
   private _uploadLengthDeferred: boolean
 
-  // Stall detector instance
-  private _stallDetector?: StallDetector
-
   constructor(file: UploadInput, options: UploadOptions) {
     // Warn about removed options from previous versions
     if ('resume' in options) {
@@ -229,9 +222,6 @@ export class BaseUpload {
       }
     }
 
-    // Setup stall detection before starting the upload
-    this._setupStallDetection()
-
     // Note: `start` does not return a Promise or await the preparation on purpose.
     // Its supposed to return immediately and start the upload in the background.
     this._prepareAndStartUpload().catch((err) => {
@@ -361,11 +351,6 @@ export class BaseUpload {
               throw new Error('tus: Expected totalSize to be set')
             }
 
-            // Update progress timestamp for the parallel upload to track stalls
-            if (upload._stallDetector) {
-              upload._stallDetector.updateProgress()
-            }
-
             this._emitProgress(totalProgress, totalSize)
           },
           // Wait until every partial upload has an upload URL, so we can add
@@ -486,11 +471,6 @@ export class BaseUpload {
     // Set the aborted flag before any `await`s, so no new requests are started.
     this._aborted = true
 
-    // Clear any stall detection
-    if (this._stallDetector) {
-      this._stallDetector.stop()
-    }
-
     // Stop any parallel partial uploads, that have been started in _startParallelUploads.
     if (this._parallelUploads != null) {
       for (const upload of this._parallelUploads) {
@@ -585,11 +565,6 @@ export class BaseUpload {
    * @api private
    */
   private _emitProgress(bytesSent: number, bytesTotal: number | null): void {
-    // Update stall detection state if progress has been made
-    if (this._stallDetector) {
-      this._stallDetector.updateProgress()
-    }
-
     if (typeof this.options.onProgress === 'function') {
       this.options.onProgress(bytesSent, bytesTotal)
     }
@@ -868,7 +843,32 @@ export class BaseUpload {
     const start = this._offset
     let end = this._offset + this.options.chunkSize
 
+    // Create stall detector for this request if stall detection is enabled and supported
+    // but don't start it yet - we'll start it after onBeforeRequest completes
+    let stallDetector: StallDetector | undefined
+    if (this.options.stallDetection?.enabled) {
+      // Only enable stall detection if the HTTP stack supports progress events
+      if (this.options.httpStack.supportsProgressEvents?.()) {
+        stallDetector = new StallDetector(
+          this.options.stallDetection,
+          this.options.httpStack,
+          (reason: string) => {
+            // Handle stall by aborting the current request and triggering retry
+            if (this._req) {
+              this._req.abort()
+            }
+            this._retryOrEmitError(new Error(`Upload stalled: ${reason}`))
+          },
+        )
+        // Don't start yet - will be started after onBeforeRequest
+      }
+    }
+
     req.setProgressHandler((bytesSent) => {
+      // Update per-request stall detector if active
+      if (stallDetector) {
+        stallDetector.updateProgress()
+      }
       this._emitProgress(start + bytesSent, this._size)
     })
 
@@ -911,22 +911,42 @@ export class BaseUpload {
     // See https://community.transloadit.com/t/how-to-abort-hanging-companion-uploads/16488/13
     const newSize = this._offset + sizeOfValue
     if (!this._uploadLengthDeferred && done && newSize !== this._size) {
+      // Clean up stall detector before throwing error
+      if (stallDetector) {
+        stallDetector.stop()
+      }
       throw new Error(
         `upload was configured with a size of ${this._size} bytes, but the source is done after ${newSize} bytes`,
       )
     }
 
-    if (value == null) {
-      return await this._sendRequest(req)
-    }
+    try {
+      let response: HttpResponse
+      if (value == null) {
+        response = await this._sendRequest(req, undefined, stallDetector)
+      } else {
+        if (
+          this.options.protocol === PROTOCOL_IETF_DRAFT_03 ||
+          this.options.protocol === PROTOCOL_IETF_DRAFT_05
+        ) {
+          req.setHeader('Upload-Complete', done ? '?1' : '?0')
+        }
+        response = await this._sendRequest(req, value, stallDetector)
+      }
 
-    if (
-      this.options.protocol === PROTOCOL_IETF_DRAFT_03 ||
-      this.options.protocol === PROTOCOL_IETF_DRAFT_05
-    ) {
-      req.setHeader('Upload-Complete', done ? '?1' : '?0')
+      // Clean up stall detector on successful completion
+      if (stallDetector) {
+        stallDetector.stop()
+      }
+
+      return response
+    } catch (error) {
+      // Clean up stall detector on error
+      if (stallDetector) {
+        stallDetector.stop()
+      }
+      throw error
     }
-    return await this._sendRequest(req, value)
   }
 
   /**
@@ -965,8 +985,6 @@ export class BaseUpload {
   private _openRequest(method: string, url: string): HttpRequest {
     const req = openRequest(method, url, this.options)
     this._req = req
-    // Store reference to this upload instance for stall detection management
-    ;(req as ExtendedHttpRequest)._upload = this
     return req
   }
 
@@ -1026,64 +1044,12 @@ export class BaseUpload {
    *
    * @api private
    */
-  _sendRequest(req: HttpRequest, body?: SliceType): Promise<HttpResponse> {
-    return sendRequest(req, body, this.options)
-  }
-
-  /**
-   * Setup stall detection monitoring
-   */
-  private _setupStallDetection(): void {
-    const stallDetection = this.options.stallDetection
-
-    // Early return if disabled or undefined
-    if (!stallDetection || !stallDetection.enabled) {
-      return
-    }
-
-    // Create stall detector instance
-    this._stallDetector = new StallDetector(
-      stallDetection,
-      this.options.httpStack,
-      (reason: string) => this._handleStall(reason),
-    )
-
-    // Start monitoring
-    this._stallDetector.start()
-  }
-
-  /**
-   * Handle a detected stall by forcing a retry
-   */
-  private _handleStall(reason: string): void {
-    // Just abort the current request, not the entire upload
-    // Each parallel upload instance has its own stall detection
-    if (this._req) {
-      this._req.abort()
-    }
-
-    // Force a retry via the error mechanism
-    this._retryOrEmitError(new Error(`Upload stalled: ${reason}`))
-  }
-
-  /**
-   * Pause stall detection temporarily
-   * @api private
-   */
-  _pauseStallDetection(): void {
-    if (this._stallDetector) {
-      this._stallDetector.pause()
-    }
-  }
-
-  /**
-   * Resume stall detection
-   * @api private
-   */
-  _resumeStallDetection(): void {
-    if (this._stallDetector) {
-      this._stallDetector.resume()
-    }
+  _sendRequest(
+    req: HttpRequest,
+    body?: SliceType,
+    stallDetector?: StallDetector,
+  ): Promise<HttpResponse> {
+    return sendRequest(req, body, this.options, stallDetector)
   }
 }
 
@@ -1144,20 +1110,15 @@ async function sendRequest(
   req: HttpRequest,
   body: SliceType | undefined,
   options: UploadOptions,
+  stallDetector?: StallDetector,
 ): Promise<HttpResponse> {
-  // Pause stall detection during onBeforeRequest callback to avoid false positives
-  const upload = (req as ExtendedHttpRequest)._upload
-  if (upload) {
-    upload._pauseStallDetection()
-  }
-
   if (typeof options.onBeforeRequest === 'function') {
     await options.onBeforeRequest(req)
   }
 
-  // Resume stall detection after callback
-  if (upload) {
-    upload._resumeStallDetection()
+  // Start stall detection after onBeforeRequest completes but before the actual network request
+  if (stallDetector) {
+    stallDetector.start()
   }
 
   const res = await req.send(body)