From a5702c0d683d71dbff0b7448722411e73b93073f Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Thu, 16 Jul 2026 14:18:13 +0000 Subject: [PATCH 1/7] feat: adding read stall support for gRPC --- storage/client_test.go | 71 ++++++++++++++++++++++++ storage/experimental/experimental.go | 7 +-- storage/grpc_client.go | 82 ++++++++++++++++++++++++---- storage/grpc_reader.go | 49 ++++++++++++++++- storage/option.go | 5 +- 5 files changed, 195 insertions(+), 19 deletions(-) diff --git a/storage/client_test.go b/storage/client_test.go index 5fc04a436f8f..ff7631c312ab 100644 --- a/storage/client_test.go +++ b/storage/client_test.go @@ -3335,6 +3335,77 @@ func TestRetryReadStallEmulated(t *testing.T) { } } +func TestGRPCRetryReadStallEmulated(t *testing.T) { + checkEmulatorEnvironment(t) + + tests := []struct { + name string + bidiReads bool + }{ + { + name: "ReadObject", + bidiReads: false, + }, + { + name: "BidiReadObject", + bidiReads: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + opts := []option.ClientOption{ + experimental.WithReadStallTimeout( + &experimental.ReadStallTimeoutConfig{ + TargetPercentile: 0.99, + Min: 10 * time.Millisecond, + }), + } + if tt.bidiReads { + opts = append(opts, experimental.WithGRPCBidiReads()) + } + + client, err := NewGRPCClient(ctx, opts...) + if err != nil { + t.Fatalf("storage.NewGRPCClient: %v", err) + } + defer client.Close() + + project := "fake-project" + bucket := fmt.Sprintf("grpc-bucket-%d", time.Now().Nanosecond()) + if err := client.Bucket(bucket).Create(ctx, project, nil); err != nil { + t.Fatalf("client.Bucket.Create: %v", err) + } + + name, _, _, err := createObjectWithContent(ctx, bucket, randomBytes3MiB) + if err != nil { + t.Fatalf("createObject: %v", err) + } + + instructions := map[string][]string{"storage.objects.get": {"stall-for-10s-after-0K"}} + testID := createRetryTest(t, client.tc, instructions) + + ctx = callctx.SetHeaders(ctx, "x-retry-test-id", testID) + r, err := client.Bucket(bucket).Object(name).NewReader(ctx) + if err != nil { + t.Fatalf("NewReader: %v", err) + } + defer r.Close() + + buf := &bytes.Buffer{} + if _, err := io.Copy(buf, r); err != nil { + t.Fatalf("io.Copy: %v", err) + } + if !bytes.Equal(buf.Bytes(), randomBytes3MiB) { + t.Errorf("content does not match, got len %v, want len %v", buf.Len(), len(randomBytes3MiB)) + } + }) + } +} + func TestWriterChunkTransferTimeoutEmulated(t *testing.T) { transportClientTest(skipGRPC("service is not implemented"), t, func(t *testing.T, ctx context.Context, project, bucket string, client storageClient) { _, err := client.CreateBucket(ctx, project, bucket, &BucketAttrs{}, nil) diff --git a/storage/experimental/experimental.go b/storage/experimental/experimental.go index dbad7b634a48..adf47bc698d8 100644 --- a/storage/experimental/experimental.go +++ b/storage/experimental/experimental.go @@ -52,7 +52,7 @@ func WithMeterProvider(mp *metric.MeterProvider) option.ClientOption { return internal.WithMeterProvider.(func(*metric.MeterProvider) option.ClientOption)(mp) } -// WithReadStallTimeout provides a [option.ClientOption] that may be passed to [storage.NewClient]. +// WithReadStallTimeout provides a [option.ClientOption] that may be passed to [storage.NewClient] or [storage.NewGRPCClient]. // It enables the client to retry stalled requests when starting a download from // Cloud Storage. If the timeout elapses with no response from the server, the request // is automatically retried. @@ -60,9 +60,8 @@ func WithMeterProvider(mp *metric.MeterProvider) option.ClientOption { // latency across all read requests from the client for each bucket accessed, and can // adjust the timeout higher to the target percentile when latency for request to that // bucket is high. -// Currently, this is supported only for downloads ([storage.NewReader] and -// [storage.NewRangeReader] calls) and only for the XML API. Other read APIs (gRPC & JSON) -// will be supported soon. +// Currently, this is supported for downloads ([storage.NewReader] and +// [storage.NewRangeReader] calls) on the XML API and gRPC API. func WithReadStallTimeout(rstc *ReadStallTimeoutConfig) option.ClientOption { return internal.WithReadStallTimeout.(func(config *ReadStallTimeoutConfig) option.ClientOption)(rstc) } diff --git a/storage/grpc_client.go b/storage/grpc_client.go index 16adf5617c5c..bdd8e8b983cf 100644 --- a/storage/grpc_client.go +++ b/storage/grpc_client.go @@ -25,10 +25,12 @@ import ( "os" "strconv" "strings" + "time" "cloud.google.com/go/iam/apiv1/iampb" gapic "cloud.google.com/go/storage/internal/apiv2" "cloud.google.com/go/storage/internal/apiv2/storagepb" + "github.com/google/uuid" "github.com/googleapis/gax-go/v2" "google.golang.org/api/iterator" "google.golang.org/api/option" @@ -120,12 +122,13 @@ func defaultGRPCOptions() []option.ClientOption { // grpcStorageClient is the gRPC API implementation of the transport-agnostic // storageClient interface. type grpcStorageClient struct { - raw *gapic.Client - settings *settings - config *storageConfig - dpDiag string - metrics *clientMetrics - metricsCleanup func() + raw *gapic.Client + settings *settings + config *storageConfig + dynamicReadReqStallTimeout *bucketDelayManager + dpDiag string + metrics *clientMetrics + metricsCleanup func() // configFeatureAttributes tracks client-level features that are enabled for this // client instance. @@ -200,11 +203,26 @@ func newGRPCStorageClient(ctx context.Context, opts ...storageOption) (client *g }() } + var bd *bucketDelayManager + if config.readStallTimeoutConfig != nil { + drrstConfig := config.readStallTimeoutConfig + bd, err = newBucketDelayManager( + drrstConfig.TargetPercentile, + getDynamicReadReqIncreaseRateFromEnv(), + getDynamicReadReqInitialTimeoutSecFromEnv(drrstConfig.Min), + drrstConfig.Min, + defaultDynamicReqdReqMaxTimeout) + if err != nil { + return nil, fmt.Errorf("creating dynamic-delay: %w", err) + } + } + c := &grpcStorageClient{ - settings: s, - config: &config, - metrics: clientMetrics, - metricsCleanup: metricsCleanup, + settings: s, + config: &config, + metrics: clientMetrics, + metricsCleanup: metricsCleanup, + dynamicReadReqStallTimeout: bd, } // Add routing interceptors to inject headers. ui, si := c.routingInterceptors() @@ -1251,6 +1269,7 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange ctx, _ = startSpan(ctx, "grpcStorageClient.NewRangeReader") defer func() { endSpan(ctx, err) }() + requestID := uuid.New() s := callSettings(c.settings, opts...) s.gax = append(s.gax, gax.WithGRPCOptions( @@ -1310,7 +1329,7 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange var err error var decoder *readResponseDecoder - err = run(cc, func(ctx context.Context) error { + openStream := func(ctx context.Context) error { var databufs mem.BufferSlice openAndSendReq := func() error { databufs = mem.BufferSlice{} @@ -1368,6 +1387,47 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange } err = decoder.readFullObjectResponse() return err + } + + err = run(cc, func(ctx context.Context) error { + if c.dynamicReadReqStallTimeout == nil { + return openStream(ctx) + } + + cancelCtx, cancel := context.WithCancel(ctx) + var ( + innerErr error + done = make(chan struct{}, 1) + ) + + go func() { + reqStartTime := time.Now() + innerErr = openStream(cancelCtx) + if innerErr == nil { + reqLatency := time.Since(reqStartTime) + c.dynamicReadReqStallTimeout.update(params.bucket, reqLatency) + } else if errors.Is(innerErr, context.Canceled) || status.Code(innerErr) == codes.Canceled { + c.dynamicReadReqStallTimeout.increase(params.bucket) + } + done <- struct{}{} + }() + + stallTimeout := c.dynamicReadReqStallTimeout.getValue(params.bucket) + timer := time.After(stallTimeout) + select { + case <-timer: + log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) + cancel() + <-done + if decoder != nil && decoder.databufs != nil { + decoder.databufs.Free() + decoder = nil + } + return context.DeadlineExceeded + case <-done: + cancel() + } + return innerErr }, s.retry, s.idempotent, withOperation("ReadObject"), withBucket(params.bucket), withObject(params.object)) if err != nil { // Close the stream context we just created to ensure we don't leak diff --git a/storage/grpc_reader.go b/storage/grpc_reader.go index 62356080aa65..a89142ebd13c 100644 --- a/storage/grpc_reader.go +++ b/storage/grpc_reader.go @@ -21,12 +21,17 @@ import ( "fmt" "hash/crc32" "io" + "log" + "time" "cloud.google.com/go/storage/internal/apiv2/storagepb" + "github.com/google/uuid" "github.com/googleapis/gax-go/v2" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/encoding" "google.golang.org/grpc/mem" + "google.golang.org/grpc/status" "google.golang.org/protobuf/encoding/protowire" "google.golang.org/protobuf/proto" ) @@ -85,6 +90,7 @@ func (c *grpcStorageClient) NewRangeReaderReadObject(ctx context.Context, params ctx, _ = startSpan(ctx, "grpcStorageClient.NewRangeReaderReadObject") defer func() { endSpan(ctx, err) }() + requestID := uuid.New() s := callSettings(c.settings, opts...) s.gax = append(s.gax, gax.WithGRPCOptions( @@ -134,7 +140,7 @@ func (c *grpcStorageClient) NewRangeReaderReadObject(ctx context.Context, params var err error var decoder *readObjectResponseDecoder - err = run(cc, func(ctx context.Context) error { + openStream := func(ctx context.Context) error { stream, err = c.raw.ReadObject(ctx, req, s.gax...) if err != nil { return err @@ -157,6 +163,47 @@ func (c *grpcStorageClient) NewRangeReaderReadObject(ctx context.Context, params } err = decoder.readFullObjectResponse() return err + } + + err = run(cc, func(ctx context.Context) error { + if c.dynamicReadReqStallTimeout == nil { + return openStream(ctx) + } + + cancelCtx, cancel := context.WithCancel(ctx) + var ( + innerErr error + done = make(chan struct{}, 1) + ) + + go func() { + reqStartTime := time.Now() + innerErr = openStream(cancelCtx) + if innerErr == nil { + reqLatency := time.Since(reqStartTime) + c.dynamicReadReqStallTimeout.update(params.bucket, reqLatency) + } else if errors.Is(innerErr, context.Canceled) || status.Code(innerErr) == codes.Canceled { + c.dynamicReadReqStallTimeout.increase(params.bucket) + } + done <- struct{}{} + }() + + stallTimeout := c.dynamicReadReqStallTimeout.getValue(params.bucket) + timer := time.After(stallTimeout) + select { + case <-timer: + log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) + cancel() + <-done + if decoder != nil && decoder.databufs != nil { + decoder.databufs.Free() + decoder = nil + } + return context.DeadlineExceeded + case <-done: + cancel() + } + return innerErr }, s.retry, s.idempotent, withOperation("ReadObject"), withBucket(params.bucket), withObject(params.object)) if err != nil { // Close the stream context we just created to ensure we don't leak diff --git a/storage/option.go b/storage/option.go index c49b224ec3ef..00ecd0db6898 100644 --- a/storage/option.go +++ b/storage/option.go @@ -243,13 +243,12 @@ func (w *withTestMetricReaderConfig) ApplyStorageOpt(c *storageConfig) { c.manualReader = w.metricReader } -// WithReadStallTimeout is an option that may be passed to [NewClient]. +// WithReadStallTimeout is an option that may be passed to [NewClient] or [NewGRPCClient]. // It enables the client to retry the stalled read request, happens as part of // storage.Reader creation. As the name suggest, timeout is adjusted dynamically // based on past observed read-req latencies. // -// This is only supported for the read operation and that too for http(XML) client. -// Grpc read-operation will be supported soon. +// This is supported for read operations on HTTP (XML) and gRPC clients. func withReadStallTimeout(rstc *experimental.ReadStallTimeoutConfig) option.ClientOption { // TODO (raj-prince): To keep separate dynamicDelay instance for different BucketHandle. // Currently, dynamicTimeout is kept at the client and hence shared across all the From 6f28be4bd80a1e1970854559d0f28626216d6f29 Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Thu, 16 Jul 2026 14:52:04 +0000 Subject: [PATCH 2/7] simplyfing for all the flow --- storage/dynamic_delay.go | 57 ++++++++++++++++++++++++++++++++++++++++ storage/grpc_client.go | 37 ++------------------------ storage/grpc_reader.go | 40 ++-------------------------- storage/http_client.go | 41 +++++------------------------ 4 files changed, 68 insertions(+), 107 deletions(-) diff --git a/storage/dynamic_delay.go b/storage/dynamic_delay.go index 5944f515d39c..8dfed5d6cc39 100644 --- a/storage/dynamic_delay.go +++ b/storage/dynamic_delay.go @@ -15,10 +15,17 @@ package storage import ( + "context" + "errors" "fmt" + "log" "math" "sync" "time" + + "github.com/google/uuid" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // dynamicDelay dynamically calculates the delay at a fixed percentile, based on @@ -235,3 +242,53 @@ func (b *bucketDelayManager) update(bucketName string, latency time.Duration) { func (b *bucketDelayManager) getValue(bucketName string) time.Duration { return b.getDelay(bucketName).getValue() } + +// executeWithReadStallTimeout executes openStream with dynamic delay stall retry tracking. +func executeWithReadStallTimeout( + ctx context.Context, + dm *bucketDelayManager, + bucket string, + requestID uuid.UUID, + openStream func(ctx context.Context) error, + onStall func(), +) error { + if dm == nil { + return openStream(ctx) + } + + cancelCtx, cancel := context.WithCancel(ctx) + var ( + innerErr error + done = make(chan struct{}, 1) + ) + + go func() { + reqStartTime := time.Now() + innerErr = openStream(cancelCtx) + if innerErr == nil { + reqLatency := time.Since(reqStartTime) + dm.update(bucket, reqLatency) + } else if (errors.Is(innerErr, context.Canceled) || status.Code(innerErr) == codes.Canceled) && ctx.Err() == nil { + dm.increase(bucket) + } + done <- struct{}{} + }() + + stallTimeout := dm.getValue(bucket) + timer := time.NewTimer(stallTimeout) + defer timer.Stop() + + select { + case <-timer.C: + log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) + cancel() + <-done + if onStall != nil { + onStall() + } + return context.DeadlineExceeded + case <-done: + cancel() + } + return innerErr +} diff --git a/storage/grpc_client.go b/storage/grpc_client.go index bdd8e8b983cf..d92cdc3d0318 100644 --- a/storage/grpc_client.go +++ b/storage/grpc_client.go @@ -25,7 +25,6 @@ import ( "os" "strconv" "strings" - "time" "cloud.google.com/go/iam/apiv1/iampb" gapic "cloud.google.com/go/storage/internal/apiv2" @@ -1390,44 +1389,12 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange } err = run(cc, func(ctx context.Context) error { - if c.dynamicReadReqStallTimeout == nil { - return openStream(ctx) - } - - cancelCtx, cancel := context.WithCancel(ctx) - var ( - innerErr error - done = make(chan struct{}, 1) - ) - - go func() { - reqStartTime := time.Now() - innerErr = openStream(cancelCtx) - if innerErr == nil { - reqLatency := time.Since(reqStartTime) - c.dynamicReadReqStallTimeout.update(params.bucket, reqLatency) - } else if errors.Is(innerErr, context.Canceled) || status.Code(innerErr) == codes.Canceled { - c.dynamicReadReqStallTimeout.increase(params.bucket) - } - done <- struct{}{} - }() - - stallTimeout := c.dynamicReadReqStallTimeout.getValue(params.bucket) - timer := time.After(stallTimeout) - select { - case <-timer: - log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) - cancel() - <-done + return executeWithReadStallTimeout(ctx, c.dynamicReadReqStallTimeout, params.bucket, requestID, openStream, func() { if decoder != nil && decoder.databufs != nil { decoder.databufs.Free() decoder = nil } - return context.DeadlineExceeded - case <-done: - cancel() - } - return innerErr + }) }, s.retry, s.idempotent, withOperation("ReadObject"), withBucket(params.bucket), withObject(params.object)) if err != nil { // Close the stream context we just created to ensure we don't leak diff --git a/storage/grpc_reader.go b/storage/grpc_reader.go index a89142ebd13c..86b8e69d5835 100644 --- a/storage/grpc_reader.go +++ b/storage/grpc_reader.go @@ -21,17 +21,13 @@ import ( "fmt" "hash/crc32" "io" - "log" - "time" "cloud.google.com/go/storage/internal/apiv2/storagepb" "github.com/google/uuid" "github.com/googleapis/gax-go/v2" "google.golang.org/grpc" - "google.golang.org/grpc/codes" "google.golang.org/grpc/encoding" "google.golang.org/grpc/mem" - "google.golang.org/grpc/status" "google.golang.org/protobuf/encoding/protowire" "google.golang.org/protobuf/proto" ) @@ -166,44 +162,12 @@ func (c *grpcStorageClient) NewRangeReaderReadObject(ctx context.Context, params } err = run(cc, func(ctx context.Context) error { - if c.dynamicReadReqStallTimeout == nil { - return openStream(ctx) - } - - cancelCtx, cancel := context.WithCancel(ctx) - var ( - innerErr error - done = make(chan struct{}, 1) - ) - - go func() { - reqStartTime := time.Now() - innerErr = openStream(cancelCtx) - if innerErr == nil { - reqLatency := time.Since(reqStartTime) - c.dynamicReadReqStallTimeout.update(params.bucket, reqLatency) - } else if errors.Is(innerErr, context.Canceled) || status.Code(innerErr) == codes.Canceled { - c.dynamicReadReqStallTimeout.increase(params.bucket) - } - done <- struct{}{} - }() - - stallTimeout := c.dynamicReadReqStallTimeout.getValue(params.bucket) - timer := time.After(stallTimeout) - select { - case <-timer: - log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) - cancel() - <-done + return executeWithReadStallTimeout(ctx, c.dynamicReadReqStallTimeout, params.bucket, requestID, openStream, func() { if decoder != nil && decoder.databufs != nil { decoder.databufs.Free() decoder = nil } - return context.DeadlineExceeded - case <-done: - cancel() - } - return innerErr + }) }, s.retry, s.idempotent, withOperation("ReadObject"), withBucket(params.bucket), withObject(params.object)) if err != nil { // Close the stream context we just created to ensure we don't leak diff --git a/storage/http_client.go b/storage/http_client.go index 629713464d5c..a6a2e3007390 100644 --- a/storage/http_client.go +++ b/storage/http_client.go @@ -21,7 +21,6 @@ import ( "fmt" "hash/crc32" "io" - "log" "net/http" "net/url" "os" @@ -1003,42 +1002,16 @@ func (c *httpStorageClient) newRangeReaderXML(ctx context.Context, params *newRa return c.hc.Do(req.WithContext(ctx)) } - cancelCtx, cancel := context.WithCancel(ctx) - var ( - res *http.Response - err error - ) - - done := make(chan bool) - go func() { - reqStartTime := time.Now() - res, err = c.hc.Do(req.WithContext(cancelCtx)) - if err == nil { - reqLatency := time.Since(reqStartTime) - c.dynamicReadReqStallTimeout.update(params.bucket, reqLatency) - } else if errors.Is(err, context.Canceled) { - // context.Canceled means operation took more than current dynamicTimeout, - // hence should be increased. - c.dynamicReadReqStallTimeout.increase(params.bucket) - } - done <- true - }() - - // Wait until stall timeout or request is successful. - stallTimeout := c.dynamicReadReqStallTimeout.getValue(params.bucket) - timer := time.After(stallTimeout) - select { - case <-timer: - log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) - cancel() - <-done + var res *http.Response + err := executeWithReadStallTimeout(ctx, c.dynamicReadReqStallTimeout, params.bucket, requestID, func(ctx context.Context) error { + var err error + res, err = c.hc.Do(req.WithContext(ctx)) + return err + }, func() { if res != nil && res.Body != nil { res.Body.Close() } - return res, context.DeadlineExceeded - case <-done: - cancel = nil - } + }) return res, err }, func() error { return setConditionsHeaders(req.Header, params.conds) }, From d954839e37636db2d051290e1900ec34acdfe7dd Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Thu, 16 Jul 2026 15:12:21 +0000 Subject: [PATCH 3/7] improving test coverage --- storage/dynamic_delay_test.go | 75 ++++++++++++++++++++++++++++ storage/experimental/experimental.go | 2 +- storage/option.go | 2 +- 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/storage/dynamic_delay_test.go b/storage/dynamic_delay_test.go index 57318580c454..24ec4d632a44 100644 --- a/storage/dynamic_delay_test.go +++ b/storage/dynamic_delay_test.go @@ -13,6 +13,8 @@ package storage import ( + "context" + "errors" "fmt" "math" "math/rand" @@ -22,6 +24,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" + "github.com/google/uuid" ) func applySamples(numSamples int, expectedValue float64, rnd *rand.Rand, d *dynamicDelay) int { @@ -383,3 +386,75 @@ func TestBucketDelayManagerMapSize(t *testing.T) { t.Errorf("Expected %d buckets in the map, but got %d", numBuckets, len(b.delays)) } } + +func TestExecuteWithReadStallTimeout(t *testing.T) { + requestID := uuid.New() + + t.Run("nil_delay_manager", func(t *testing.T) { + executed := false + err := executeWithReadStallTimeout(context.Background(), nil, "bucket", requestID, func(ctx context.Context) error { + executed = true + return nil + }, nil) + if err != nil { + t.Errorf("expected nil error, got %v", err) + } + if !executed { + t.Error("expected openStream to be executed") + } + }) + + t.Run("fast_success", func(t *testing.T) { + dm, _ := newBucketDelayManager(0.99, 1.5, 100*time.Millisecond, 10*time.Millisecond, 10*time.Second) + err := executeWithReadStallTimeout(context.Background(), dm, "bucket", requestID, func(ctx context.Context) error { + time.Sleep(2 * time.Millisecond) + return nil + }, nil) + if err != nil { + t.Errorf("expected nil error, got %v", err) + } + }) + + t.Run("stall_timeout_triggered", func(t *testing.T) { + dm, _ := newBucketDelayManager(0.99, 1.5, 10*time.Millisecond, 10*time.Millisecond, 10*time.Second) + initialVal := dm.getValue("bucket") + stalled := false + + err := executeWithReadStallTimeout(context.Background(), dm, "bucket", requestID, func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() + }, func() { + stalled = true + }) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("expected context.DeadlineExceeded, got %v", err) + } + if !stalled { + t.Error("expected onStall callback to be invoked") + } + if newVal := dm.getValue("bucket"); newVal <= initialVal { + t.Errorf("expected dynamic timeout to increase, initial %v, new %v", initialVal, newVal) + } + }) + + t.Run("outer_context_cancelled", func(t *testing.T) { + dm, _ := newBucketDelayManager(0.99, 1.5, 100*time.Millisecond, 10*time.Millisecond, 10*time.Second) + initialVal := dm.getValue("bucket") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel outer context immediately. + + err := executeWithReadStallTimeout(ctx, dm, "bucket", requestID, func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() + }, nil) + + if !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got %v", err) + } + if newVal := dm.getValue("bucket"); newVal != initialVal { + t.Errorf("expected dynamic timeout NOT to increase on outer context cancellation, initial %v, new %v", initialVal, newVal) + } + }) +} diff --git a/storage/experimental/experimental.go b/storage/experimental/experimental.go index adf47bc698d8..5c6136c3fea9 100644 --- a/storage/experimental/experimental.go +++ b/storage/experimental/experimental.go @@ -61,7 +61,7 @@ func WithMeterProvider(mp *metric.MeterProvider) option.ClientOption { // adjust the timeout higher to the target percentile when latency for request to that // bucket is high. // Currently, this is supported for downloads ([storage.NewReader] and -// [storage.NewRangeReader] calls) on the XML API and gRPC API. +// [storage.NewRangeReader] calls) on the XML API and gRPC API (not implemented for JSON API). func WithReadStallTimeout(rstc *ReadStallTimeoutConfig) option.ClientOption { return internal.WithReadStallTimeout.(func(config *ReadStallTimeoutConfig) option.ClientOption)(rstc) } diff --git a/storage/option.go b/storage/option.go index 00ecd0db6898..41eb5d2e395c 100644 --- a/storage/option.go +++ b/storage/option.go @@ -248,7 +248,7 @@ func (w *withTestMetricReaderConfig) ApplyStorageOpt(c *storageConfig) { // storage.Reader creation. As the name suggest, timeout is adjusted dynamically // based on past observed read-req latencies. // -// This is supported for read operations on HTTP (XML) and gRPC clients. +// This is supported for read operations on HTTP (XML) and gRPC clients (not implemented for HTTP JSON client). func withReadStallTimeout(rstc *experimental.ReadStallTimeoutConfig) option.ClientOption { // TODO (raj-prince): To keep separate dynamicDelay instance for different BucketHandle. // Currently, dynamicTimeout is kept at the client and hence shared across all the From 8e6c4e58f77be5d42fa449e8d0a38ef4185c4993 Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Mon, 20 Jul 2026 18:23:20 +0000 Subject: [PATCH 4/7] resolving merge conflict --- storage/client_test.go | 1 + storage/option.go | 6 ------ 2 files changed, 1 insertion(+), 6 deletions(-) diff --git a/storage/client_test.go b/storage/client_test.go index ff7631c312ab..51cbaaff8dbb 100644 --- a/storage/client_test.go +++ b/storage/client_test.go @@ -3373,6 +3373,7 @@ func TestGRPCRetryReadStallEmulated(t *testing.T) { t.Fatalf("storage.NewGRPCClient: %v", err) } defer client.Close() + client.SetRetry(WithBackoff(gax.Backoff{Initial: 10 * time.Millisecond})) project := "fake-project" bucket := fmt.Sprintf("grpc-bucket-%d", time.Now().Nanosecond()) diff --git a/storage/option.go b/storage/option.go index 41eb5d2e395c..d058f574e817 100644 --- a/storage/option.go +++ b/storage/option.go @@ -250,12 +250,6 @@ func (w *withTestMetricReaderConfig) ApplyStorageOpt(c *storageConfig) { // // This is supported for read operations on HTTP (XML) and gRPC clients (not implemented for HTTP JSON client). func withReadStallTimeout(rstc *experimental.ReadStallTimeoutConfig) option.ClientOption { - // TODO (raj-prince): To keep separate dynamicDelay instance for different BucketHandle. - // Currently, dynamicTimeout is kept at the client and hence shared across all the - // BucketHandle, which is not the ideal state. As latency depends on location of VM - // and Bucket, and read latency of different buckets may lie in different range. - // Hence having a separate dynamicTimeout instance at BucketHandle level will - // be better if rstc.Min == time.Duration(0) { rstc.Min = defaultDynamicReadReqMinTimeout } From a523105504d089dfe3b80b498727ef8eee0e6440 Mon Sep 17 00:00:00 2001 From: Prince Kumar Date: Mon, 20 Jul 2026 18:59:00 +0000 Subject: [PATCH 5/7] fix(storage): deduplicate x-goog-api-client header in setHeadersFromCtx and adjust emulator read stall test parameters --- storage/client_test.go | 4 ++-- storage/http_client.go | 18 +++++++++++------- 2 files changed, 13 insertions(+), 9 deletions(-) diff --git a/storage/client_test.go b/storage/client_test.go index 51cbaaff8dbb..019ef3ccc270 100644 --- a/storage/client_test.go +++ b/storage/client_test.go @@ -3295,7 +3295,7 @@ func TestRetryReadStallEmulated(t *testing.T) { client, err := NewClient(ctx, experimental.WithReadStallTimeout( &experimental.ReadStallTimeoutConfig{ TargetPercentile: 0.99, - Min: 10 * time.Millisecond, + Min: 250 * time.Millisecond, })) if err != nil { t.Fatalf("storage.NewClient: %v", err) @@ -3361,7 +3361,7 @@ func TestGRPCRetryReadStallEmulated(t *testing.T) { experimental.WithReadStallTimeout( &experimental.ReadStallTimeoutConfig{ TargetPercentile: 0.99, - Min: 10 * time.Millisecond, + Min: 250 * time.Millisecond, }), } if tt.bidiReads { diff --git a/storage/http_client.go b/storage/http_client.go index a6a2e3007390..fa012accf004 100644 --- a/storage/http_client.go +++ b/storage/http_client.go @@ -1736,14 +1736,18 @@ func setHeadersFromCtx(ctx context.Context, header http.Header) { // Merge x-goog-api-client values into a single space-separated value. if strings.EqualFold(k, xGoogHeaderKey) { alreadySetValues := header.Values(xGoogHeaderKey) - vals = append(vals, alreadySetValues...) - - if len(vals) > 0 { - xGoogHeader := vals[0] - for _, v := range vals[1:] { - xGoogHeader = strings.Join([]string{xGoogHeader, v}, " ") + var uniqueVals []string + seenVals := make(map[string]bool) + for _, v := range append(vals, alreadySetValues...) { + for _, part := range strings.Split(v, " ") { + if part != "" && !seenVals[part] { + seenVals[part] = true + uniqueVals = append(uniqueVals, part) + } } - header.Set(k, xGoogHeader) + } + if len(uniqueVals) > 0 { + header.Set(k, strings.Join(uniqueVals, " ")) } } else { for _, v := range vals { From 0bfa9329a273487477d520fd1a476de2f455fb5d Mon Sep 17 00:00:00 2001 From: raj-prince Date: Tue, 21 Jul 2026 12:19:37 +0000 Subject: [PATCH 6/7] scoping down to legacy grpc read path --- storage/client_test.go | 101 ++++++++++++++++----------------------- storage/dynamic_delay.go | 9 ++-- 2 files changed, 45 insertions(+), 65 deletions(-) diff --git a/storage/client_test.go b/storage/client_test.go index 019ef3ccc270..beb80b1014c9 100644 --- a/storage/client_test.go +++ b/storage/client_test.go @@ -3337,74 +3337,55 @@ func TestRetryReadStallEmulated(t *testing.T) { func TestGRPCRetryReadStallEmulated(t *testing.T) { checkEmulatorEnvironment(t) + t.Run("ReadObject", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() - tests := []struct { - name string - bidiReads bool - }{ - { - name: "ReadObject", - bidiReads: false, - }, - { - name: "BidiReadObject", - bidiReads: true, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() + opts := []option.ClientOption{ + experimental.WithReadStallTimeout( + &experimental.ReadStallTimeoutConfig{ + TargetPercentile: 0.99, + Min: 250 * time.Millisecond, + }), + } - opts := []option.ClientOption{ - experimental.WithReadStallTimeout( - &experimental.ReadStallTimeoutConfig{ - TargetPercentile: 0.99, - Min: 250 * time.Millisecond, - }), - } - if tt.bidiReads { - opts = append(opts, experimental.WithGRPCBidiReads()) - } + client, err := NewGRPCClient(ctx, opts...) + if err != nil { + t.Fatalf("storage.NewGRPCClient: %v", err) + } + defer client.Close() + client.SetRetry(WithBackoff(gax.Backoff{Initial: 10 * time.Millisecond})) - client, err := NewGRPCClient(ctx, opts...) - if err != nil { - t.Fatalf("storage.NewGRPCClient: %v", err) - } - defer client.Close() - client.SetRetry(WithBackoff(gax.Backoff{Initial: 10 * time.Millisecond})) + project := "fake-project" + bucket := fmt.Sprintf("grpc-bucket-%d", time.Now().Nanosecond()) + if err := client.Bucket(bucket).Create(ctx, project, nil); err != nil { + t.Fatalf("client.Bucket.Create: %v", err) + } - project := "fake-project" - bucket := fmt.Sprintf("grpc-bucket-%d", time.Now().Nanosecond()) - if err := client.Bucket(bucket).Create(ctx, project, nil); err != nil { - t.Fatalf("client.Bucket.Create: %v", err) - } + name, _, _, err := createObjectWithContent(ctx, bucket, randomBytes3MiB) + if err != nil { + t.Fatalf("createObject: %v", err) + } - name, _, _, err := createObjectWithContent(ctx, bucket, randomBytes3MiB) - if err != nil { - t.Fatalf("createObject: %v", err) - } + instructions := map[string][]string{"storage.objects.get": {"stall-for-10s-after-0K"}} + testID := createRetryTest(t, client.tc, instructions) - instructions := map[string][]string{"storage.objects.get": {"stall-for-10s-after-0K"}} - testID := createRetryTest(t, client.tc, instructions) + testCtx := callctx.SetHeaders(ctx, "x-retry-test-id", testID) - ctx = callctx.SetHeaders(ctx, "x-retry-test-id", testID) - r, err := client.Bucket(bucket).Object(name).NewReader(ctx) - if err != nil { - t.Fatalf("NewReader: %v", err) - } - defer r.Close() + r, err := client.Bucket(bucket).Object(name).NewReader(testCtx) + if err != nil { + t.Fatalf("NewReader: %v", err) + } + defer r.Close() - buf := &bytes.Buffer{} - if _, err := io.Copy(buf, r); err != nil { - t.Fatalf("io.Copy: %v", err) - } - if !bytes.Equal(buf.Bytes(), randomBytes3MiB) { - t.Errorf("content does not match, got len %v, want len %v", buf.Len(), len(randomBytes3MiB)) - } - }) - } + buf := &bytes.Buffer{} + if _, err := io.Copy(buf, r); err != nil { + t.Fatalf("io.Copy: %v", err) + } + if !bytes.Equal(buf.Bytes(), randomBytes3MiB) { + t.Errorf("content does not match, got len %v, want len %v", buf.Len(), len(randomBytes3MiB)) + } + }) } func TestWriterChunkTransferTimeoutEmulated(t *testing.T) { diff --git a/storage/dynamic_delay.go b/storage/dynamic_delay.go index 8dfed5d6cc39..47943dc55413 100644 --- a/storage/dynamic_delay.go +++ b/storage/dynamic_delay.go @@ -16,7 +16,6 @@ package storage import ( "context" - "errors" "fmt" "log" "math" @@ -24,8 +23,6 @@ import ( "time" "github.com/google/uuid" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) // dynamicDelay dynamically calculates the delay at a fixed percentile, based on @@ -268,7 +265,7 @@ func executeWithReadStallTimeout( if innerErr == nil { reqLatency := time.Since(reqStartTime) dm.update(bucket, reqLatency) - } else if (errors.Is(innerErr, context.Canceled) || status.Code(innerErr) == codes.Canceled) && ctx.Err() == nil { + } else if (ctx.Err() == nil && cancelCtx.Err() != nil) { dm.increase(bucket) } done <- struct{}{} @@ -288,7 +285,9 @@ func executeWithReadStallTimeout( } return context.DeadlineExceeded case <-done: - cancel() + if innerErr != nil { + cancel() + } } return innerErr } From 80ea3f41167df9f56bb18bdfd4b4a84a750dd95a Mon Sep 17 00:00:00 2001 From: raj-prince Date: Tue, 21 Jul 2026 12:30:40 +0000 Subject: [PATCH 7/7] vet and rebase --- storage/dynamic_delay.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/storage/dynamic_delay.go b/storage/dynamic_delay.go index 47943dc55413..63c68e84215b 100644 --- a/storage/dynamic_delay.go +++ b/storage/dynamic_delay.go @@ -265,7 +265,7 @@ func executeWithReadStallTimeout( if innerErr == nil { reqLatency := time.Since(reqStartTime) dm.update(bucket, reqLatency) - } else if (ctx.Err() == nil && cancelCtx.Err() != nil) { + } else if ctx.Err() == nil && cancelCtx.Err() != nil { dm.increase(bucket) } done <- struct{}{}