fix: do not re-use readstreams when retry an upload of the test repla…

…y recording (#29745)

* custom async retry for any async fn

* "proxied" fetch for PUT, throwing http errors

* cleans up put_fetch return types & parsing

* move protocol upload logic to put_protocol_artifact, rm upload_stream

* changelog

* changelog

* manually fix wsp in snapshot

* reorganizes a little, properly tests retryable logic

* removes 408 as a non-retriable error in unit test

* make expectation to be resolved more clear

* rm unnecessary type coercion

* changelog

* Update cli/CHANGELOG.md

* asyncRetry only throws Aggregate errors if there is more than one attempt

* Update packages/server/test/unit/cloud/network/is_retryable_error_spec.ts

Co-authored-by: Matt Schile <mschile@cypress.io>

* Update packages/server/test/unit/cloud/network/is_retryable_error_spec.ts

---------

Co-authored-by: Matt Schile <mschile@cypress.io>

cli/CHANGELOG.md

@@ -1,4 +1,12 @@
 <!-- See the ../guides/writing-the-cypress-changelog.md for details on writing the changelog. -->
+## 13.13.1
+
+_Released 7/16/2024 (PENDING)_
+
+**Bugfixes:**
+
+- Fixed an issue where the ReadStream used to upload a Test Replay recording could erroneously be re-used when retrying in cases of retryable upload failures. Fixes [#29227](https://github.com/cypress-io/cypress/issues/29227)
+
 ## 13.13.0
 
 _Released 7/01/2024_

packages/server/lib/cloud/api/put_protocol_artifact.ts

@@ -1,9 +1,10 @@
 import fsAsync from 'fs/promises'
 import fs from 'fs'
 import Debug from 'debug'
-import { uploadStream, geometricRetry } from '../upload/upload_stream'
 import { StreamActivityMonitor } from '../upload/stream_activity_monitor'
-
+import { asyncRetry, linearDelay } from '../../util/async_retry'
+import { putFetch, ParseKinds } from '../network/put_fetch'
+import { isRetryableError } from '../network/is_retryable_error'
 const debug = Debug('cypress:server:cloud:api:protocol-artifact')
 
 // the upload will get canceled if the source stream does not
@@ -13,23 +14,37 @@ const debug = Debug('cypress:server:cloud:api:protocol-artifact')
 const MAX_START_DWELL_TIME = 5000
 const MAX_ACTIVITY_DWELL_TIME = 5000
 
-export const putProtocolArtifact = async (artifactPath: string, maxFileSize: number, destinationUrl: string) => {
-  debug(`Atttempting to upload Test Replay archive from ${artifactPath} to ${destinationUrl})`)
-  const { size } = await fsAsync.stat(artifactPath)
+export const _delay = linearDelay(500)
+
+export const putProtocolArtifact = asyncRetry(
+  async (artifactPath: string, maxFileSize: number, destinationUrl: string) => {
+    debug(`Atttempting to upload Test Replay archive from ${artifactPath} to ${destinationUrl})`)
+    const { size } = await fsAsync.stat(artifactPath)
 
-  if (size > maxFileSize) {
-    throw new Error(`Spec recording too large: artifact is ${size} bytes, limit is ${maxFileSize} bytes`)
-  }
+    if (size > maxFileSize) {
+      throw new Error(`Spec recording too large: artifact is ${size} bytes, limit is ${maxFileSize} bytes`)
+    }
 
-  const activityMonitor = new StreamActivityMonitor(MAX_START_DWELL_TIME, MAX_ACTIVITY_DWELL_TIME)
-  const fileStream = fs.createReadStream(artifactPath)
+    const activityMonitor = new StreamActivityMonitor(MAX_START_DWELL_TIME, MAX_ACTIVITY_DWELL_TIME)
+    const fileStream = fs.createReadStream(artifactPath)
+    const controller = activityMonitor.getController()
 
-  await uploadStream(
-    fileStream,
-    destinationUrl,
-    size, {
-      retryDelay: geometricRetry,
-      activityMonitor,
-    },
-  )
-}
+    await putFetch(destinationUrl, {
+      parse: ParseKinds.TEXT,
+      headers: {
+        'content-length': String(size),
+        'content-type': 'application/x-tar',
+        'accept': 'application/json',
+      },
+      // ts thinks this is a web fetch, which only expects ReadableStreams.
+      // But, this is a node fetch, which supports ReadStreams.
+      // @ts-expect-error
+      body: activityMonitor.monitor(fileStream),
+      signal: controller.signal,
+    })
+  }, {
+    maxAttempts: 3,
+    retryDelay: _delay,
+    shouldRetry: isRetryableError,
+  },
+)

packages/server/lib/cloud/artifacts/upload_artifacts.ts

@@ -11,8 +11,8 @@ import { IArtifact, ArtifactUploadResult, ArtifactKinds } from './artifact'
 import { createScreenshotArtifactBatch } from './screenshot_artifact'
 import { createVideoArtifact } from './video_artifact'
 import { createProtocolArtifact, composeProtocolErrorReportFromOptions } from './protocol_artifact'
-import { HttpError } from '../api/http_error'
-import { NetworkError } from '../api/network_error'
+import { HttpError } from '../network/http_error'
+import { NetworkError } from '../network/network_error'
 
 const debug = Debug('cypress:server:cloud:artifacts')
 
@@ -216,6 +216,7 @@ export const uploadArtifacts = async (options: UploadArtifactOptions) => {
     if (postUploadProtocolFatalError && postUploadProtocolFatalError.captureMethod === 'uploadCaptureArtifact') {
       const error = postUploadProtocolFatalError.error
 
+      debug('protocol error: %O', error)
       if ((error as AggregateError).errors) {
         // eslint-disable-next-line no-console
         console.log('')

packages/server/lib/cloud/api/http_error.ts → packages/server/lib/cloud/network/http_error.ts

@@ -1,4 +1,4 @@
-import { scrubUrl } from './scrub_url'
+import { scrubUrl } from '../api/scrub_url'
 
 export const HttpErrorKind = 'HttpError'
 

packages/server/lib/cloud/network/is_retryable_error.ts

@@ -0,0 +1,9 @@
+import { NetworkError } from './network_error'
+import { HttpError } from './http_error'
+
+export const isRetryableError = (error?: Error) => {
+  return error ? (
+    NetworkError.isNetworkError(error) ||
+    HttpError.isHttpError(error) && [408, 429, 502, 503, 504].includes(error.status)
+  ) : false
+}

packages/server/lib/cloud/network/parse_error.ts

@@ -0,0 +1,11 @@
+const ParseErrorKind = 'ParseErrorKind'
+
+export class ParseError extends Error {
+  public readonly kind = ParseErrorKind
+  constructor (public readonly originalError: Error, message?: string) {
+    super(message)
+  }
+  static isParseError (err: Error & { kind: string }): err is ParseError {
+    return err.kind === ParseErrorKind
+  }
+}

packages/server/lib/cloud/network/put_fetch.ts

@@ -0,0 +1,82 @@
+import crossFetch from 'cross-fetch'
+import { NetworkError } from './network_error'
+import { HttpError } from './http_error'
+import { ParseError } from './parse_error'
+import { agent } from '@packages/network'
+import Debug from 'debug'
+
+const debug = Debug('cypress-verbose:server:cloud:api:put')
+
+type PutInit = Omit<RequestInit, 'agent' | 'method'>
+
+export const ParseKinds = Object.freeze({
+  JSON: 'json',
+  TEXT: 'text',
+})
+
+type ParseKind = typeof ParseKinds[keyof typeof ParseKinds]
+
+type PutOptions = PutInit & {
+  parse?: ParseKind
+}
+
+export async function putFetch <
+  TReturn extends any
+> (input: RequestInfo | URL, options: PutOptions = { parse: 'json' }): Promise<TReturn> {
+  const {
+    parse,
+    ...init
+  } = options
+
+  debug('Initiating PUT %s', input)
+  try {
+    const response = await crossFetch(input, {
+      ...(init || {}),
+      method: 'PUT',
+      // cross-fetch thinks this is in the browser, so declares
+      // types based on that rather than on node-fetch which it
+      // actually uses under the hood. node-fetch supports `agent`.
+      // @ts-expect-error
+      agent,
+    })
+
+    if (response.status >= 400) {
+      const err = await HttpError.fromResponse(response)
+
+      throw err
+    }
+
+    try {
+      switch (parse) {
+        case ParseKinds.JSON:
+          return await response.json() as TReturn
+        case ParseKinds.TEXT:
+          return await response.text() as TReturn
+        default:
+          return response.body as any
+      }
+    } catch (e) {
+      const parseError = new ParseError(e, e.message)
+
+      parseError.stack = e.stack
+      throw parseError
+    }
+  } catch (e) {
+    debug('Error: %O', e)
+    if (ParseError.isParseError(e)) {
+      throw e
+    } else if (HttpError.isHttpError(e)) {
+      throw e
+    }
+
+    // if the error wasn't a parsing error, it's probably a Network error
+    const url = typeof input === 'string' ? input :
+      input instanceof URL ? input.href :
+        input instanceof Request ? input.url : 'UNKNOWN_URL'
+
+    const networkError = new NetworkError(e, url)
+
+    networkError.stack = e.stack
+    throw networkError
+  }
+}