feat(storage): configurable retries for uploads (#5210)

Uploads will now use the configured retry strategy for the client, taking advantage of the new surface in the apiary library for this. Only idempotent uploads will be retried by default.

storage/emulator_test.sh

@@ -67,62 +67,5 @@ function cleanup() {
 }
 trap cleanup EXIT
 
-# TODO: move to passing once fixed
-FAILING=(   
-            "objects.insert"
-        )
-# TODO: remove regex once all tests are passing
-# Unfortunately, there is no simple way to skip specific tests (see https://github.com/golang/go/issues/41583)
-# Therefore, we have to simply run all the specific tests we know pass
-PASSING=(   "buckets.list"
-            "buckets.insert"
-            "buckets.get"
-            "buckets.delete"
-            "buckets.update"
-            "buckets.patch"
-            "buckets.getIamPolicy"
-            "buckets.setIamPolicy"
-            "buckets.testIamPermissions"
-            "buckets.lockRetentionPolicy"
-            "objects.copy"
-            "objects.get"
-            "objects.list"
-            "objects.delete"
-            "objects.update"
-            "objects.patch"
-            "objects.compose"
-            "objects.rewrite"
-            "serviceaccount.get"
-            "hmacKey.get"
-            "hmacKey.list"
-            "hmacKey.create"
-            "hmacKey.delete"
-            "hmacKey.update"
-            "notifications.list"
-            "notifications.create"
-            "notifications.get"
-            "notifications.delete"
-            "object_acl.insert"
-            "object_acl.get"
-            "object_acl.list"
-            "object_acl.patch"
-            "object_acl.update"
-            "object_acl.delete"
-            "default_object_acl.insert"
-            "default_object_acl.get"
-            "default_object_acl.list"
-            "default_object_acl.patch"
-            "default_object_acl.update"
-            "default_object_acl.delete"
-            "bucket_acl.insert"
-            "bucket_acl.get"
-            "bucket_acl.list"
-            "bucket_acl.patch"
-            "bucket_acl.update"
-            "bucket_acl.delete"
-        )
-TEMP=${PASSING[@]} 
-PASSING_REGEX=${TEMP// /|}
-
 # Run tests
-go test -v -timeout 10m ./ -run="TestRetryConformance/($PASSING_REGEX)-" -short 2>&1 | tee -a sponge_log.log
+go test -v -timeout 10m ./ -run="TestRetryConformance" -short 2>&1 | tee -a sponge_log.log

storage/writer.go

@@ -172,6 +172,22 @@ func (w *Writer) open() error {
 			// call to set up the upload as well as calls to upload individual chunks
 			// for a resumable upload (as long as the chunk size is non-zero). Hence
 			// there is no need to add retries here.
+
+			// Retry only when the operation is idempotent or the retry policy is RetryAlways.
+			isIdempotent := w.o.conds != nil && (w.o.conds.GenerationMatch >= 0 || w.o.conds.DoesNotExist == true)
+			var useRetry bool
+			if (w.o.retry == nil || w.o.retry.policy == RetryIdempotent) && isIdempotent {
+				useRetry = true
+			} else if w.o.retry != nil && w.o.retry.policy == RetryAlways {
+				useRetry = true
+			}
+			if useRetry {
+				if w.o.retry != nil {
+					call.WithRetry(w.o.retry.backoff, w.o.retry.shouldRetry)
+				} else {
+					call.WithRetry(nil, nil)
+				}
+			}
 			resp, err = call.Do()
 		}
 		if err != nil {

storage/writer_test.go

@@ -37,7 +37,7 @@ func TestErrorOnObjectsInsertCall(t *testing.T) {
 
 	doWrite := func(mt *mockTransport) *Writer {
 		client := mockClient(t, mt)
-		wc := client.Bucket("bucketname").Object("filename1").NewWriter(ctx)
+		wc := client.Bucket("bucketname").Object("filename1").If(Conditions{DoesNotExist: true}).NewWriter(ctx)
 		wc.ContentType = "text/plain"
 
 		// We can't check that the Write fails, since it depends on the write to the