feat(storage): configurable retries for uploads

Depends on merge and release of googleapis/google-api-go-client#1324

storage/writer.go

@@ -172,6 +172,25 @@ 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.
+			var isIdempotent bool
+			if w.o.conds != nil && (w.o.conds.GenerationMatch >= 0 || w.o.conds.DoesNotExist == true) {
+				isIdempotent = 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

@@ -36,7 +36,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