From b43ff8a706732f69b2a3623331931d2f069a74bf Mon Sep 17 00:00:00 2001 From: Sneha Sahu Date: Wed, 15 Jul 2026 08:41:46 +0000 Subject: [PATCH] storage: fix retry fallback in HTTP OpenWriter In http_client.go, pass a unified retryFunc callback into call.WithRetry to enforce storage.ShouldRetry fallback and safely handle nil retry configs. In client_test.go, set googleapi.DefaultUploadChunkSize on openWriterParams in TestWriterRetryMaxAttemptsEmulated so media chunk buffering enables HTTP retries, and capture synchronous pw.Close() error propagation to correctly assert max attempt errors. --- storage/client_test.go | 327 ++++++++++++++++++++++++++++++ storage/http_client.go | 38 +++- storage/retry_conformance_test.go | 84 ++++++++ 3 files changed, 440 insertions(+), 9 deletions(-) diff --git a/storage/client_test.go b/storage/client_test.go index 5fc04a436f8f..4930ef8abf98 100644 --- a/storage/client_test.go +++ b/storage/client_test.go @@ -40,12 +40,15 @@ import ( "github.com/googleapis/gax-go/v2" "github.com/googleapis/gax-go/v2/apierror" "github.com/googleapis/gax-go/v2/callctx" + "google.golang.org/api/googleapi" "google.golang.org/api/iterator" "google.golang.org/api/option" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" expgrpc "google.golang.org/grpc/experimental" "google.golang.org/grpc/mem" + "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" ) @@ -3243,6 +3246,62 @@ func TestRetryMaxDurationEmulated(t *testing.T) { }) } +// Test that writer errors are wrapped correctly if retry happens until max attempts. +func TestWriterRetryMaxAttemptsEmulated(t *testing.T) { + transportClientTest(context.Background(), t, func(t *testing.T, ctx context.Context, project, bucket string, client storageClient) { + _, err := client.CreateBucket(ctx, project, bucket, &BucketAttrs{}, nil) + if err != nil { + t.Fatalf("creating bucket: %v", err) + } + instructions := map[string][]string{"storage.objects.insert": {"return-503", "return-503", "return-503", "return-503", "return-503"}} + testID := createRetryTest(t, client, instructions) + ctx = callctx.SetHeaders(ctx, "x-retry-test-id", testID) + config := &retryConfig{maxAttempts: intPointer(3), backoff: &gax.Backoff{Initial: 10 * time.Millisecond}} + + prefix := time.Now().Nanosecond() + want := &ObjectAttrs{ + Bucket: bucket, + Name: fmt.Sprintf("%d-object-%d", prefix, time.Now().Nanosecond()), + Generation: defaultGen, + } + + var writeErr error + params := &openWriterParams{ + attrs: want, + bucket: bucket, + chunkSize: googleapi.DefaultUploadChunkSize, + ctx: ctx, + donec: make(chan struct{}), + setError: func(err error) { writeErr = err }, + progress: func(_ int64) {}, // no-op + setObj: func(o *ObjectAttrs) {}, + setSize: func(int64) {}, + } + + pw, err := client.OpenWriter(params, idempotent(true), withRetryConfig(config)) + if err != nil { + t.Fatalf("failed to open writer: %v", err) + } + if _, err := pw.Write([]byte("A")); err != nil { + t.Fatalf("failed to write data: %v", err) + } + if err := pw.Close(); err != nil && writeErr == nil { + writeErr = err + } + <-params.donec + + var ae *apierror.APIError + if errors.As(writeErr, &ae) { + if ae.GRPCStatus().Code() != codes.Unavailable && ae.HTTPCode() != 503 { + t.Errorf("OpenWriter: got unexpected error %v; want 503", writeErr) + } + } + if got, want := writeErr.Error(), "retry failed after 3 attempts"; !strings.Contains(got, want) { + t.Errorf("got error: %q, want to contain: %q", got, want) + } + }) +} + // Test that a timeout returns a DeadlineExceeded error, in spite of DeadlineExceeded being a retryable // status when it is returned by the server. func TestTimeoutErrorEmulated(t *testing.T) { @@ -4004,3 +4063,271 @@ func setBidiReads(t *testing.T, client storageClient) { }) } } + +func TestClient_RetryAndHeadersHTTP(t *testing.T) { + ctx := context.Background() + + cases := []struct { + name string + call func(ctx context.Context, client *Client) error + wantAttempts int + }{ + { + name: "Attrs", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").Attrs(ctx) + return err + }, + wantAttempts: 2, + }, + { + name: "Delete", + call: func(ctx context.Context, client *Client) error { + return client.Bucket("b").Object("o").Delete(ctx) + }, + wantAttempts: 2, + }, + { + name: "Update", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").Update(ctx, ObjectAttrsToUpdate{}) + return err + }, + wantAttempts: 2, + }, + { + name: "Restore", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").Restore(ctx, &RestoreOptions{}) + return err + }, + wantAttempts: 2, + }, + { + name: "CopyTo", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").CopierFrom(client.Bucket("b").Object("o2")).Run(ctx) + return err + }, + wantAttempts: 2, + }, + { + name: "ComposerFrom", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").ComposerFrom(client.Bucket("b").Object("o2")).Run(ctx) + return err + }, + wantAttempts: 2, + }, + { + name: "NewReader", + call: func(ctx context.Context, client *Client) error { + r, err := client.Bucket("b").Object("o").NewReader(ctx) + if err == nil { + r.Close() + } + return err + }, + wantAttempts: 2, + }, + { + name: "NewWriter", + call: func(ctx context.Context, client *Client) error { + w := client.Bucket("b").Object("o").NewWriter(ctx) + if _, err := w.Write([]byte("data")); err != nil { + w.Close() + return err + } + return w.Close() + }, + wantAttempts: 2, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + mt := &mockTransport{} + // Add enough results for the expected attempts plus one spare + for i := 0; i < tc.wantAttempts+1; i++ { + mt.addResult(&http.Response{StatusCode: 503, Body: bodyReader("Unavailable")}, nil) + } + + client := mockClient(t, mt) + client.SetRetry( + WithMaxAttempts(2), + WithBackoff(gax.Backoff{Initial: time.Millisecond}), + WithPolicy(RetryAlways), + ) + + err := tc.call(ctx, client) + if err == nil { + t.Fatalf("expected error, got nil") + } + + if got, want := len(mt.results), 1; got != want { + t.Errorf("got remaining results %d, want %d", got, want) + } + + req := mt.gotReq + if req == nil { + t.Fatalf("no request captured") + } + apiClientHeader := req.Header.Get("X-Goog-Api-Client") + if !strings.Contains(apiClientHeader, "gccl-invocation-id/") { + t.Errorf("missing gccl-invocation-id in header: %q", apiClientHeader) + } + wantAttemptStr := fmt.Sprintf("gccl-attempt-count/%d", tc.wantAttempts) + if !strings.Contains(apiClientHeader, wantAttemptStr) { + t.Errorf("missing/incorrect gccl-attempt-count in header (expected %d): %q", tc.wantAttempts, apiClientHeader) + } + }) + } +} + +func TestClient_RetryAndHeadersGRPC(t *testing.T) { + ctx := context.Background() + + cases := []struct { + name string + call func(ctx context.Context, client *Client) error + }{ + { + name: "Attrs", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").Attrs(ctx) + return err + }, + }, + { + name: "Delete", + call: func(ctx context.Context, client *Client) error { + return client.Bucket("b").Object("o").Delete(ctx) + }, + }, + { + name: "Update", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").Update(ctx, ObjectAttrsToUpdate{}) + return err + }, + }, + { + name: "Restore", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").Restore(ctx, &RestoreOptions{}) + return err + }, + }, + { + name: "CopyTo", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").CopierFrom(client.Bucket("b").Object("o2")).Run(ctx) + return err + }, + }, + { + name: "ComposerFrom", + call: func(ctx context.Context, client *Client) error { + _, err := client.Bucket("b").Object("o").ComposerFrom(client.Bucket("b").Object("o2")).Run(ctx) + return err + }, + }, + { + name: "NewReader", + call: func(ctx context.Context, client *Client) error { + r, err := client.Bucket("b").Object("o").NewReader(ctx) + if err == nil { + r.Close() + } + return err + }, + }, + { + name: "NewWriter", + call: func(ctx context.Context, client *Client) error { + w := client.Bucket("b").Object("o").NewWriter(ctx) + if _, err := w.Write([]byte("data")); err != nil { + w.Close() + return err + } + return w.Close() + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var gotAttempts int + var gotMetadata []metadata.MD + var mu sync.Mutex + + unaryInterceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { + mu.Lock() + gotAttempts++ + if md, ok := metadata.FromOutgoingContext(ctx); ok { + gotMetadata = append(gotMetadata, md) + } + mu.Unlock() + return status.Error(codes.Unavailable, "Unavailable") + } + + streamInterceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { + mu.Lock() + gotAttempts++ + if md, ok := metadata.FromOutgoingContext(ctx); ok { + gotMetadata = append(gotMetadata, md) + } + mu.Unlock() + return nil, status.Error(codes.Unavailable, "Unavailable") + } + + client, err := NewGRPCClient(ctx, + option.WithoutAuthentication(), + option.WithGRPCDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())), + option.WithGRPCDialOption(grpc.WithUnaryInterceptor(unaryInterceptor)), + option.WithGRPCDialOption(grpc.WithStreamInterceptor(streamInterceptor)), + ) + if err != nil { + t.Fatalf("failed to create gRPC client: %v", err) + } + defer client.Close() + + client.SetRetry( + WithMaxAttempts(2), + WithBackoff(gax.Backoff{Initial: time.Millisecond}), + WithPolicy(RetryAlways), + ) + + err = tc.call(ctx, client) + if err == nil { + t.Fatalf("expected error, got nil") + } + + mu.Lock() + attempts := gotAttempts + mds := gotMetadata + mu.Unlock() + + if attempts != 2 { + t.Errorf("got attempts %d, want 2", attempts) + } + + if len(mds) < 2 { + t.Fatalf("got metadata length %d, want at least 2", len(mds)) + } + + lastMD := mds[len(mds)-1] + clientHeaderList := lastMD.Get("x-goog-api-client") + if len(clientHeaderList) == 0 { + t.Fatalf("missing x-goog-api-client header in metadata") + } + clientHeader := clientHeaderList[0] + if !strings.Contains(clientHeader, "gccl-invocation-id/") { + t.Errorf("missing gccl-invocation-id in header: %q", clientHeader) + } + if !strings.Contains(clientHeader, "gccl-attempt-count/2") { + t.Errorf("missing/incorrect gccl-attempt-count in header (expected 2): %q", clientHeader) + } + }) + } +} diff --git a/storage/http_client.go b/storage/http_client.go index b9f387437869..1819dba710cc 100644 --- a/storage/http_client.go +++ b/storage/http_client.go @@ -34,6 +34,7 @@ import ( "cloud.google.com/go/iam/apiv1/iampb" "cloud.google.com/go/internal/optional" "github.com/google/uuid" + gax "github.com/googleapis/gax-go/v2" "github.com/googleapis/gax-go/v2/callctx" "google.golang.org/api/googleapi" "google.golang.org/api/iterator" @@ -1204,21 +1205,40 @@ func (c *httpStorageClient) OpenWriter(params *openWriterParams, opts ...storage } else if s.retry != nil && s.retry.policy == RetryAlways { useRetry = true } + attempts := 1 + var maxAttemptsReached bool if useRetry { + var bo *gax.Backoff if s.retry != nil { - // Wrap shouldRetry to adapt to the googleapi WithRetry signature. - var retryFunc func(error) bool - if s.retry.shouldRetry != nil { - retryFunc = func(err error) bool { - return s.retry.shouldRetry(err, nil) - } + bo = s.retry.backoff + } + + // Wrap shouldRetry to adapt to the googleapi WithRetry signature. + retryFunc := func(err error) bool { + if s.retry != nil && s.retry.maxAttempts != nil && attempts >= *s.retry.maxAttempts { + maxAttemptsReached = true + return false + } + + var retryable bool + if s.retry != nil && s.retry.shouldRetry != nil { + retryable = s.retry.shouldRetry(err, nil) + } else { + retryable = ShouldRetry(err) } - call.WithRetry(s.retry.backoff, retryFunc) - } else { - call.WithRetry(nil, nil) + if retryable { + attempts++ + } + return retryable } + call.WithRetry(bo, retryFunc) } resp, err = call.Do() + if err != nil && s.retry != nil { + if maxAttemptsReached && s.retry.maxAttempts != nil { + err = fmt.Errorf("storage: retry failed after %v attempts; last error: %w", *s.retry.maxAttempts, err) + } + } } if err != nil { errorf(err) diff --git a/storage/retry_conformance_test.go b/storage/retry_conformance_test.go index 78f529e093b4..8be8b6339cea 100644 --- a/storage/retry_conformance_test.go +++ b/storage/retry_conformance_test.go @@ -1352,3 +1352,87 @@ func (et *emulatorTest) delete() { et.Errorf("deleting test: err: %v, resp: %+v", err, resp) } } + +func TestRetryConformance_MaxAttempts(t *testing.T) { + host := os.Getenv("STORAGE_EMULATOR_HOST") + if host == "" { + t.Skip("This test must use the testbench emulator; set STORAGE_EMULATOR_HOST to run.") + } + endpoint, err := url.Parse(host) + if err != nil { + t.Fatalf("error parsing emulator host: %v", err) + } + + ctx := context.Background() + client, err := NewClient(ctx) + if err != nil { + t.Fatalf("storage.NewClient: %v", err) + } + + skippedGRPCMethods := []string{"storage.hmacKey.delete", "storage.hmacKey.create", "storage.hmacKey.list", "storage.hmacKey.get", "storage.hmacKey.update", "storage.serviceaccount.get"} + skippedHTTPMethods := []string{"storage.appendable.upload"} + + _, _, testFiles := parseFiles(t) + + for _, testFile := range testFiles { + for _, retryTest := range testFile.RetryTests { + if !retryTest.ExpectSuccess { + continue + } + for _, instructions := range retryTest.Cases { + if len(instructions.Instructions) < 2 { + continue + } + for _, method := range retryTest.Methods { + methodName := method.Name + if method.Group != "" { + methodName = method.Group + } + if len(methods[methodName]) == 0 { + continue + } + + for i, fn := range methods[methodName] { + transports := []string{"http", "grpc"} + for _, transport := range transports { + testName := fmt.Sprintf("%v-%v-%v-%v-%v", transport, retryTest.Id, instructions.Instructions, methodName, i) + t.Run(testName, func(t *testing.T) { + if transport == "http" && slices.Contains(skippedHTTPMethods, methodName) { + t.Skip("not supported") + } + if transport == "grpc" && slices.Contains(skippedGRPCMethods, methodName) { + t.Skip("not supported") + } + + subtest := &emulatorTest{T: t, name: testName, host: endpoint} + subtest.create(map[string][]string{ + method.Name: instructions.Instructions, + }, transport) + defer subtest.delete() + + subtest.populateResources(ctx, client, method.Resources) + + maxAttempts := 1 + subtest.transportClient.SetRetry( + WithMaxAttempts(maxAttempts), + WithBackoff(gax.Backoff{Initial: 10 * time.Millisecond}), + WithPolicy(RetryAlways), + ) + + ctx := callctx.SetHeaders(context.Background(), "x-retry-test-id", subtest.id) + err = fn(ctx, subtest.transportClient, &subtest.resources, retryTest.PreconditionProvided) + if err == nil { + t.Fatalf("expected failure due to max attempts exceeded, got success") + } + expectedSubstr := fmt.Sprintf("retry failed after %d attempts", maxAttempts) + if !strings.Contains(err.Error(), expectedSubstr) { + t.Errorf("got error %q, want error to contain %q", err.Error(), expectedSubstr) + } + }) + } + } + } + } + } + } +}