fix(storage): follow-ups for read stall retry

[cpriti-os]

Follow-ups to #20525.

1. XML reads: don't leak invocation headers across retries.
newRangeReaderXML reuses one *http.Request for every attempt, and setHeadersFromCtx merges x-goog-api-client values into whatever the header already contains. As a result, retried requests carried the attempt counts from all earlier attempts, e.g. on attempt 3:
gccl-invocation-id/… gccl-attempt-count/3 gl-go/… gccl/… gccl-attempt-count/2 gccl-attempt-count/1.
The fix snapshots the header value before the first attempt and restores it at the start of each attempt, before the per-attempt context headers are applied. The new test TestXMLReaderRetryInvocationHeaders checks that each of 3 attempts carries exactly one gccl-attempt-count/N (the correct N) and exactly one gccl-invocation-id.

2. Emulator stall tests: cover BidiReadObject, and assert the stall was injected.

• TestGRPCRetryReadStallEmulated now runs for both ReadObject and BidiReadObject (WithGRPCBidiReads()).
• The HTTP and gRPC stall tests now call a new helper, checkRetryTestCompleted, which reuses emulatorTest.check(). It fails the test if the testbench didn't use the stall instruction, so the tests can no longer pass without a stall ever happening.
• Requires storage-testbench v0.64.0 (gRPC read stall support, feat: Add gRPC read stall support for Go SDK storage-testbench#819).

3. Stall metric description.
The description of gcp.storage.client.stall.duration said "waiting for the first byte", which is only accurate for HTTP. It now says: stall timeout after which a read attempt was aborted while waiting for the initial response (response headers for HTTP, first response message for gRPC).

4. executeWithReadStallTimeout: prefer the real result when it arrives at the same moment the timer fires.
If openStream finished just as the stall timer fired, select could pick the timer case at random. That threw away a real success (or a real error such as NotFound) and reported a stall instead. The timer case now first checks whether openStream has already finished and, if so, returns its result.

Testing

• go build ./..., go vet ., gofmt -l .: clean.
• go test -short -race -count=1 .: pass.
• go test -short -race -count=50 -run TestExecuteWithReadStallTimeout .: pass.
• TestXMLReaderRetryInvocationHeaders fails without fix 1 (2 or 3 gccl-attempt-count tokens on attempts 2 and 3) and passes with it.
• Emulator tests against storage-testbench v0.64.0: TestRetryReadStallEmulated, TestGRPCRetryReadStallEmulated/ReadObject and TestGRPCRetryReadStallEmulated/BidiReadObject all pass (~0.3s each: stall, then retry).
• The same tests against v0.63.0 fail with test not completed; unused instructions: map[storage.objects.get:[stall-for-10s-after-0K]], which confirms the new assertion works.

[gemini-code-assist]

Code Review

This pull request introduces several improvements and fixes to the storage client. In storage/dynamic_delay.go, it prevents false-positive stall reports by checking if the stream completed at the same time the timer fired. In storage/http_client.go, it snapshots and restores the x-goog-api-client header to prevent attempt-specific tokens from leaking across retries. It also updates tests to cover both standard and bidirectional gRPC reads, and adds a test verifying XML reader retry invocation headers. Feedback on the changes highlights a potential issue in storage/dynamic_delay.go where setting cancel = nil to prevent deferred cancellation may not work if defer cancel() is evaluated immediately rather than wrapped in a closure. Additionally, it is recommended to use a long-lived context for NewGRPCClient in tests to avoid premature cancellation of background tasks, and to use r.Header.Values instead of r.Header.Get in tests to thoroughly assert against leaked headers.

[gemini-code-assist]

Setting cancel = nil to prevent deferred cancellation only works if the deferred cancel is wrapped in a closure (e.g., defer func() { if cancel != nil { cancel() } }()). If the function instead uses a direct defer cancel(), the function value is evaluated immediately at the defer statement, and the original cancel function will still execute upon return, prematurely canceling the successful stream. If the direct defer cancel() pattern is used elsewhere in this function, it should be refactored to a closure.

[cpriti-os]

The deferred cancel is already a closure (defer func() { if cancel != nil { cancel() } }(), L253-257), so setting cancel = nil prevents it from running. No change needed.

[gemini-code-assist]

Passing a short-lived context ctx (with a 5-second timeout) to the client constructor NewGRPCClient violates the best practice of using a long-lived context for client initialization. Since the client may spawn background goroutines for connection monitoring and session maintenance, a short-lived context risks premature cancellation of these background tasks once the timeout expires. Passing a long-lived context like context.Background() to NewGRPCClient instead, and restricting the use of the timeout context ctx to individual RPC calls, avoids this issue.

Suggested change

	client, err := NewGRPCClient(ctx, opts...)
	client, err := NewGRPCClient(context.Background(), opts...)

References

1. When initializing clients or channels (constructors) that may spawn background goroutines (e.g., for session maintenance or connection monitoring), pass a long-lived context directly instead of a short-lived or bounded context. Do not wrap constructors in short-lived timeout helpers if the constructor itself doesn't use that timeout context, as it provides no hang protection and risks premature cancellation of background tasks. Instead, apply bounded timeouts only to discrete RPCs.

[cpriti-os]

The context only covers client setup and the test's RPCs, and client.Close() runs before cancel(), so nothing outlives it. Keeping it consistent with TestRetryReadStallEmulated.

[gemini-code-assist]

Using r.Header.Get(xGoogHeaderKey) only retrieves the first value of the header. If the client implementation leaks headers by appending them as separate header fields (via Header.Add), those leaked values will reside in subsequent elements of the header slice and will be silently ignored by Header.Get. To thoroughly assert against any leaked headers, retrieve all values using r.Header.Values(xGoogHeaderKey) and join them.

Suggested change

	gotHeaders = append(gotHeaders, r.Header.Get(xGoogHeaderKey))
	gotHeaders = append(gotHeaders, strings.Join(r.Header.Values(xGoogHeaderKey), " "))

[cpriti-os]

Good catch, switched to strings.Join(r.Header.Values(...), " ").

[krishnamd-jkp]

nit: Consider extracting this to a single place to avoid missing changes in both the locations in the future.

[cpriti-os]

Done