feat(storage): add retry config to IAM handle and reader

storage/emulator_test.sh

@@ -56,7 +56,7 @@ function cleanup() {
 trap cleanup EXIT
 
 # TODO: move to passing once fixed
-FAILING=(   "buckets.setIamPolicy"
+FAILING=(   
             "objects.insert"
         )
 # TODO: remove regex once all tests are passing
@@ -69,6 +69,7 @@ PASSING=(   "buckets.list"
             "buckets.update"
             "buckets.patch"
             "buckets.getIamPolicy"
+            "buckets.setIamPolicy"
             "buckets.testIamPermissions"
             "buckets.lockRetentionPolicy"
             "objects.copy"

storage/iam.go

@@ -29,13 +29,15 @@ func (b *BucketHandle) IAM() *iam.Handle {
 	return iam.InternalNewHandleClient(&iamClient{
 		raw:         b.c.raw,
 		userProject: b.userProject,
+		retry:       b.retry,
 	}, b.name)
 }
 
 // iamClient implements the iam.client interface.
 type iamClient struct {
 	raw         *raw.Service
 	userProject string
+	retry       *retryConfig
 }
 
 func (c *iamClient) Get(ctx context.Context, resource string) (p *iampb.Policy, err error) {
@@ -52,10 +54,10 @@ func (c *iamClient) GetWithVersion(ctx context.Context, resource string, request
 		call.UserProject(c.userProject)
 	}
 	var rp *raw.Policy
-	err = runWithRetry(ctx, func() error {
+	err = run(ctx, func() error {
 		rp, err = call.Context(ctx).Do()
 		return err
-	})
+	}, c.retry, true)
 	if err != nil {
 		return nil, err
 	}
@@ -72,10 +74,11 @@ func (c *iamClient) Set(ctx context.Context, resource string, p *iampb.Policy) (
 	if c.userProject != "" {
 		call.UserProject(c.userProject)
 	}
-	return runWithRetry(ctx, func() error {
+	isIdempotent := len(p.Etag) > 0
+	return run(ctx, func() error {
 		_, err := call.Context(ctx).Do()
 		return err
-	})
+	}, c.retry, isIdempotent)
 }
 
 func (c *iamClient) Test(ctx context.Context, resource string, perms []string) (permissions []string, err error) {
@@ -88,10 +91,10 @@ func (c *iamClient) Test(ctx context.Context, resource string, perms []string) (
 		call.UserProject(c.userProject)
 	}
 	var res *raw.TestIamPermissionsResponse
-	err = runWithRetry(ctx, func() error {
+	err = run(ctx, func() error {
 		res, err = call.Context(ctx).Do()
 		return err
-	})
+	}, c.retry, true)
 	if err != nil {
 		return nil, err
 	}

storage/reader.go

@@ -163,7 +163,7 @@ func (o *ObjectHandle) NewRangeReader(ctx context.Context, offset, length int64)
 		}
 
 		var res *http.Response
-		err = runWithRetry(ctx, func() error {
+		err = run(ctx, func() error {
 			res, err = o.c.hc.Do(req)
 			if err != nil {
 				return err
@@ -210,7 +210,7 @@ func (o *ObjectHandle) NewRangeReader(ctx context.Context, offset, length int64)
 				gen = gen64
 			}
 			return nil
-		})
+		}, o.retry, true)
 		if err != nil {
 			return nil, err
 		}
@@ -483,7 +483,7 @@ func (o *ObjectHandle) newRangeReaderWithGRPC(ctx context.Context, offset, lengt
 		var msg *storagepb.ReadObjectResponse
 		var err error
 
-		err = runWithRetry(cc, func() error {
+		err = run(cc, func() error {
 			stream, err = o.c.gc.ReadObject(cc, req)
 			if err != nil {
 				return err
@@ -492,7 +492,7 @@ func (o *ObjectHandle) newRangeReaderWithGRPC(ctx context.Context, offset, lengt
 			msg, err = stream.Recv()
 
 			return err
-		})
+		}, o.retry, true)
 		if err != nil {
 			// Close the stream context we just created to ensure we don't leak
 			// resources.

storage/retry_conformance_test.go

@@ -243,10 +243,11 @@ var methods = map[string][]retryFunc{
 				return err
 			}
 
-			if err := bkt.IAM().SetPolicy(ctx, policy); err != nil {
-				return err
+			if !preconditions {
+				policy.InternalProto.Etag = nil
 			}
-			return fmt.Errorf("Etag preconditions not supported")
+
+			return bkt.IAM().SetPolicy(ctx, policy)
 		},
 	},
 	"storage.hmacKey.update": {