Fix data race (#492)

* Fix data race

We were seeing intermittent data race failures in our unit tests. This backoff implementation is advertised as not thread safe so we need a new instance each time. https://github.com/cenkalti/backoff/blob/v4.2.0/tries.go#L10-L12

* fix remaining occurrences

* extract into functions

clients/arweave_ipfs_s3.go

@@ -16,8 +16,6 @@ import (
 const SCHEME_IPFS = "ipfs"
 const SCHEME_ARWEAVE = "ar"
 
-var dstorageRetryBackoff = backoff.WithMaxRetries(backoff.NewConstantBackOff(1*time.Second), 2)
-
 func CopyDStorageToS3(url, s3URL string, requestID string) error {
 	return backoff.Retry(func() error {
 		content, err := DownloadDStorageFromGatewayList(url, requestID)
@@ -31,7 +29,7 @@ func CopyDStorageToS3(url, s3URL string, requestID string) error {
 		}
 
 		return nil
-	}, dstorageRetryBackoff)
+	}, DStorageRetryBackoff())
 }
 
 func DownloadDStorageFromGatewayList(u string, requestID string) (io.ReadCloser, error) {
@@ -128,3 +126,7 @@ func parseDStorageGatewayURL(u *url.URL) (string, string, string) {
 
 	return "", "", ""
 }
+
+func DStorageRetryBackoff() backoff.BackOff {
+	return backoff.WithMaxRetries(backoff.NewConstantBackOff(1*time.Second), 2)
+}

clients/input_copy.go

@@ -20,8 +20,6 @@ const MAX_COPY_FILE_DURATION = 30 * time.Minute
 const MaxInputFileSizeBytes = 10 * 1024 * 1024 * 1024 // 10 GiB
 const PresignDuration = 10 * time.Minute
 
-var RETRY_BACKOFF = backoff.WithMaxRetries(newExponentialBackOffExecutor(), 5)
-
 type InputCopier interface {
 	CopyInputToS3(requestID string, inputFile, osTransferURL *url.URL) (video.InputVideo, string, error)
 }
@@ -115,7 +113,7 @@ func CopyFile(ctx context.Context, sourceURL, destOSBaseURL, filename, requestID
 		content := io.TeeReader(c, &byteAccWriter)
 
 		return UploadToOSURL(destOSBaseURL, filename, content, MAX_COPY_FILE_DURATION)
-	}, RETRY_BACKOFF)
+	}, UploadRetryBackoff())
 	return
 }
 

clients/object_store_client.go

@@ -131,3 +131,7 @@ func newExponentialBackOffExecutor() *backoff.ExponentialBackOff {
 
 	return backOff
 }
+
+func UploadRetryBackoff() backoff.BackOff {
+	return backoff.WithMaxRetries(newExponentialBackOffExecutor(), 5)
+}

transcode/manifest.go

@@ -28,7 +28,7 @@ func DownloadRenditionManifest(sourceManifestOSURL string) (m3u8.MediaPlaylist,
 			return fmt.Errorf("error decoding manifest: %s", err)
 		}
 		return nil
-	}, transcodeRetryBackoff)
+	}, TranscodeRetryBackoff())
 	if err != nil {
 		return m3u8.MediaPlaylist{}, err
 	}
@@ -138,7 +138,7 @@ func GenerateAndUploadManifests(sourceManifest m3u8.MediaPlaylist, targetOSURL s
 		renditionManifestBaseURL := fmt.Sprintf("%s/%s", targetOSURL, profile.Name)
 		err = backoff.Retry(func() error {
 			return clients.UploadToOSURL(renditionManifestBaseURL, manifestFilename, strings.NewReader(renditionPlaylist.String()), UPLOAD_TIMEOUT)
-		}, clients.RETRY_BACKOFF)
+		}, clients.UploadRetryBackoff())
 		if err != nil {
 			return "", fmt.Errorf("failed to upload rendition playlist: %s", err)
 		}
@@ -152,7 +152,7 @@ func GenerateAndUploadManifests(sourceManifest m3u8.MediaPlaylist, targetOSURL s
 
 	err := backoff.Retry(func() error {
 		return clients.UploadToOSURL(targetOSURL, MASTER_MANIFEST_FILENAME, strings.NewReader(masterPlaylist.String()), UPLOAD_TIMEOUT)
-	}, clients.RETRY_BACKOFF)
+	}, clients.UploadRetryBackoff())
 	if err != nil {
 		return "", fmt.Errorf("failed to upload master playlist: %s", err)
 	}

transcode/transcode.go

@@ -19,8 +19,6 @@ import (
 
 const UPLOAD_TIMEOUT = 5 * time.Minute
 
-var transcodeRetryBackoff = backoff.WithMaxRetries(backoff.NewConstantBackOff(5*time.Second), 10)
-
 type TranscodeSegmentRequest struct {
 	SourceFile        string                 `json:"source_location"`
 	CallbackURL       string                 `json:"callback_url"`
@@ -188,7 +186,7 @@ func transcodeSegment(
 			}
 		}
 		return nil
-	}, transcodeRetryBackoff)
+	}, TranscodeRetryBackoff())
 
 	if err != nil {
 		return err
@@ -210,7 +208,7 @@ func transcodeSegment(
 
 		err = backoff.Retry(func() error {
 			return clients.UploadToOSURL(targetRenditionURL, fmt.Sprintf("%d.ts", segment.Index), bytes.NewReader(transcodedSegment.MediaData), UPLOAD_TIMEOUT)
-		}, clients.RETRY_BACKOFF)
+		}, clients.UploadRetryBackoff())
 		if err != nil {
 			return fmt.Errorf("failed to upload master playlist: %s", err)
 		}
@@ -276,3 +274,7 @@ type RenditionStats struct {
 	ManifestLocation string
 	BitsPerSecond    uint32
 }
+
+func TranscodeRetryBackoff() backoff.BackOff {
+	return backoff.WithMaxRetries(backoff.NewConstantBackOff(5*time.Second), 10)
+}