Skip to content

fix(storage): apply TransferStallTimeoutOption deadline to unary gRPC calls #16600

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ajayky-os merged 3 commits into from
Oct 29, 2026
Merged

fix(storage): apply TransferStallTimeoutOption deadline to unary gRPC calls #16600

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2214596
fix(storage): apply TransferStallTimeoutOption deadline to unary gRPC…
ajayky-os Oct 28, 2026
87d664e
style(storage): format unary deadline files and use explicit types
ajayky-os Oct 28, 2026
1ab65b6
test(storage): expand unary deadline tests and update docs per review
ajayky-os Oct 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions google/cloud/storage/internal/async/connection_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,6 +241,7 @@ future<StatusOr<google::storage::v2::Bucket>> AsyncConnectionImpl::GetBucket(
std::shared_ptr<grpc::ClientContext> context,
google::cloud::internal::ImmutableOptions options,
google::storage::v2::GetBucketRequest const& request) {
ApplyUnaryRpcDeadline(*context, *options);
return stub->AsyncGetBucket(cq, std::move(context), std::move(options),
request);
};
Expand Down Expand Up @@ -623,6 +624,7 @@ AsyncConnectionImpl::ComposeObject(ComposeObjectParams p) {
std::shared_ptr<grpc::ClientContext> context,
google::cloud::internal::ImmutableOptions options,
google::storage::v2::ComposeObjectRequest const& request) {
ApplyUnaryRpcDeadline(*context, *options);
AddIdempotencyToken(*context, id);
return stub->AsyncComposeObject(cq, std::move(context), std::move(options),
request);
Expand All @@ -646,6 +648,7 @@ future<Status> AsyncConnectionImpl::DeleteObject(DeleteObjectParams p) {
CompletionQueue& cq, std::shared_ptr<grpc::ClientContext> context,
google::cloud::internal::ImmutableOptions options,
google::storage::v2::DeleteObjectRequest const& proto) {
ApplyUnaryRpcDeadline(*context, *options);
AddIdempotencyToken(*context, id);
return stub->AsyncDeleteObject(cq, std::move(context),
std::move(options), proto);
Expand Down Expand Up @@ -756,6 +759,7 @@ AsyncConnectionImpl::StartResumableWrite(
CompletionQueue& cq, std::shared_ptr<grpc::ClientContext> context,
google::cloud::internal::ImmutableOptions options,
google::storage::v2::StartResumableWriteRequest const& proto) {
ApplyUnaryRpcDeadline(*context, *options);
Comment thread
ajayky-os marked this conversation as resolved.
internal::ConfigureContext(*context, *options);
return stub->AsyncStartResumableWrite(cq, std::move(context),
std::move(options), proto);
Expand All @@ -778,6 +782,7 @@ AsyncConnectionImpl::QueryWriteStatus(
CompletionQueue& cq, std::shared_ptr<grpc::ClientContext> context,
google::cloud::internal::ImmutableOptions options,
google::storage::v2::QueryWriteStatusRequest const& proto) {
ApplyUnaryRpcDeadline(*context, *options);
google::cloud::internal::ConfigurePollContext(*context, *options);
return stub->AsyncQueryWriteStatus(cq, std::move(context),
std::move(options), proto);
Expand Down
6 changes: 6 additions & 0 deletions google/cloud/storage/internal/async/connection_impl_test.cc
View file
Open in desktop
Comment thread
ajayky-os marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,8 @@ TEST_F(AsyncConnectionImplTest, GetBucket) {
EXPECT_THAT(GetMetadata(*context),
testing::Not(testing::Contains(
testing::Key("x-goog-gcs-idempotency-token"))));
EXPECT_LT(context->deadline(),
std::chrono::system_clock::time_point::max());
EXPECT_EQ(options->get<AuthorityOption>(), kAuthority);
auto expected = google::storage::v2::GetBucketRequest{};
EXPECT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &expected));
Expand Down Expand Up @@ -170,6 +172,8 @@ TEST_F(AsyncConnectionImplTest, ComposeObject) {
GetMetadata(*context),
testing::Contains(testing::Pair("x-goog-gcs-idempotency-token",
testing::Not(testing::IsEmpty()))));
EXPECT_LT(context->deadline(),
std::chrono::system_clock::time_point::max());
// Verify at least one option is initialized with the correct value.
EXPECT_EQ(options->get<AuthorityOption>(), kAuthority);
auto expected = google::storage::v2::ComposeObjectRequest{};
Expand Down Expand Up @@ -275,6 +279,8 @@ TEST_F(AsyncConnectionImplTest, DeleteObject) {
GetMetadata(*context),
testing::Contains(testing::Pair("x-goog-gcs-idempotency-token",
testing::Not(testing::IsEmpty()))));
EXPECT_LT(context->deadline(),
std::chrono::system_clock::time_point::max());
// Verify at least one option is initialized with the correct values.
EXPECT_EQ(options->get<AuthorityOption>(), kAuthority);
google::storage::v2::DeleteObjectRequest expected;
Expand Down
29 changes: 20 additions & 9 deletions google/cloud/storage/internal/async/connection_impl_upload_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,8 +130,10 @@ TEST_F(AsyncConnectionImplTest, StartUnbufferedUpload) {
});
})
.WillOnce(
[&](auto&, auto, auto,
[&](auto&, auto const& context, auto,
google::storage::v2::StartResumableWriteRequest const& request) {
EXPECT_EQ(context->deadline(),
std::chrono::system_clock::time_point::max());
auto expected = google::storage::v2::StartResumableWriteRequest{};
EXPECT_TRUE(
TextFormat::ParseFromString(kExpectedRequest, &expected));
Expand Down Expand Up @@ -284,13 +286,20 @@ TEST_F(AsyncConnectionImplTest, StartUnbufferedUpload) {
TEST_F(AsyncConnectionImplTest, UnbufferedUploadNewUploadWithTimeout) {
AsyncSequencer<bool> sequencer;
auto mock = std::make_shared<storage::testing::MockStorageStub>();
EXPECT_CALL(*mock, AsyncStartResumableWrite).WillOnce([&sequencer] {
return sequencer.PushBack("StartResumableWrite").then([](auto) {
auto response = google::storage::v2::StartResumableWriteResponse{};
response.set_upload_id("test-upload-id");
return make_status_or(response);
});
});
EXPECT_CALL(*mock, AsyncStartResumableWrite)
.WillOnce([&sequencer](
CompletionQueue&,
std::shared_ptr<grpc::ClientContext> const& context,
google::cloud::internal::ImmutableOptions const&,
google::storage::v2::StartResumableWriteRequest const&) {
EXPECT_LT(context->deadline(),
std::chrono::system_clock::time_point::max());
return sequencer.PushBack("StartResumableWrite").then([](auto) {
auto response = google::storage::v2::StartResumableWriteResponse{};
response.set_upload_id("test-upload-id");
return make_status_or(response);
});
});
EXPECT_CALL(*mock, AsyncBidiWriteObject).WillOnce([&]() {
auto stream = std::make_unique<MockAsyncBidiWriteObjectStream>();
EXPECT_CALL(*stream, Start).WillOnce([&sequencer] {
Expand Down Expand Up @@ -498,8 +507,10 @@ TEST_F(AsyncConnectionImplTest, ResumeUnbufferedUpload) {
});
})
.WillOnce(
[&](auto&, auto, auto,
[&](auto&, auto const& context, auto,
google::storage::v2::QueryWriteStatusRequest const& request) {
EXPECT_LT(context->deadline(),
std::chrono::system_clock::time_point::max());
auto expected = google::storage::v2::QueryWriteStatusRequest{};
EXPECT_TRUE(TextFormat::ParseFromString(kRequestText, &expected));
EXPECT_THAT(request, IsProtoEqual(expected));
Expand Down
1 change: 1 addition & 0 deletions google/cloud/storage/internal/async/rewriter_connection_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ RewriterConnectionImpl::Iterate() {
std::shared_ptr<grpc::ClientContext> context,
google::cloud::internal::ImmutableOptions options,
google::storage::v2::RewriteObjectRequest const& proto) {
ApplyUnaryRpcDeadline(*context, *options);
AddIdempotencyToken(*context, id);
return stub->AsyncRewriteObject(cq, std::move(context),
std::move(options), proto);
Expand Down
4 changes: 4 additions & 0 deletions google/cloud/storage/internal/async/rewriter_connection_impl_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,6 +249,8 @@ TEST(RewriterConnectionImplTest, IterateReusesIdempotencyTokenOnRetry) {
EXPECT_CALL(*mock, AsyncRewriteObject)
.WillOnce([&](auto&, auto const& context, auto const&,
google::storage::v2::RewriteObjectRequest const&) {
EXPECT_LT(context->deadline(),
std::chrono::system_clock::time_point::max());
auto metadata = validate_metadata_fixture.GetMetadata(*context);
auto l = metadata.find("x-goog-gcs-idempotency-token");
EXPECT_NE(l, metadata.end());
Expand All @@ -263,6 +265,8 @@ TEST(RewriterConnectionImplTest, IterateReusesIdempotencyTokenOnRetry) {
})
.WillOnce([&](auto&, auto const& context, auto const&,
google::storage::v2::RewriteObjectRequest const&) {
EXPECT_LT(context->deadline(),
std::chrono::system_clock::time_point::max());
auto metadata = validate_metadata_fixture.GetMetadata(*context);
auto l = metadata.find("x-goog-gcs-idempotency-token");
EXPECT_NE(l, metadata.end());
Expand Down
78 changes: 67 additions & 11 deletions google/cloud/storage/internal/grpc/configure_client_context.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,14 @@

#include "google/cloud/storage/internal/generic_request.h"
#include "google/cloud/storage/internal/object_requests.h"
#include "google/cloud/storage/options.h"
#include "google/cloud/storage/version.h"
#include "google/cloud/grpc_options.h"
#include "google/cloud/internal/rest_context.h"
#include "google/storage/v2/storage.pb.h"
#include <grpcpp/client_context.h>
#include <chrono>
#include <type_traits>

namespace google {
namespace cloud {
Expand All @@ -40,18 +43,36 @@ void AddIdempotencyToken(grpc::ClientContext& ctx,
/// Configures @p ctx using @p token.
void AddIdempotencyToken(grpc::ClientContext& ctx, std::string const& token);

/**
* Inject request query parameters into grpc::ClientContext.
*
* The REST API has a number of "standard" query parameters that are not part of
* the gRPC request body, instead they are sent via metadata headers in the gRPC
* request.
*
* @see https://cloud.google.com/apis/docs/system-parameters
*/
template <typename Request>
void ApplyQueryParameters(grpc::ClientContext& ctx, Options const& options,
Request const& request) {
struct IsStreamingRequest : std::false_type {};

template <>
struct IsStreamingRequest<storage::internal::ReadObjectRangeRequest>
: std::true_type {};

template <>
struct IsStreamingRequest<storage::internal::InsertObjectMediaRequest>
: std::true_type {};

template <>
struct IsStreamingRequest<storage::internal::UploadChunkRequest>
: std::true_type {};

/// Applies the unary RPC deadline from `TransferStallTimeoutOption` to @p ctx
/// if a deadline has not already been set on @p ctx.
inline void ApplyUnaryRpcDeadline(grpc::ClientContext& ctx,
Options const& options) {
std::chrono::seconds const timeout =
options.get<storage::TransferStallTimeoutOption>();
if (timeout > std::chrono::seconds::zero() &&
ctx.deadline() == std::chrono::system_clock::time_point::max()) {
ctx.set_deadline(std::chrono::system_clock::now() + timeout);
}
}

template <typename Request>
void ApplyQuotaUserAndFieldMask(grpc::ClientContext& ctx,
Request const& request, std::true_type) {
// The gRPC API has a single field for the `QuotaUser` parameter, while the
// JSON API has two:
// https://cloud.google.com/storage/docs/json_api/v1/parameters#quotaUser
Expand All @@ -71,6 +92,41 @@ void ApplyQueryParameters(grpc::ClientContext& ctx, Options const& options,
ctx.AddMetadata("x-goog-fieldmask",
request.template GetOption<storage::Fields>().value());
}
}

template <typename Request>
void ApplyQuotaUserAndFieldMask(grpc::ClientContext&, Request const&,
std::false_type) {}

/**
Comment thread
ajayky-os marked this conversation as resolved.
* Configure a grpc::ClientContext for a storage request.
*
* Inject the request query parameters into the context. The REST API has a
* number of "standard" query parameters that are not part of the gRPC request
* body, instead they are sent via metadata headers in the gRPC request.
*
* For non-streaming requests this also sets a per-attempt deadline from
* `storage::TransferStallTimeoutOption` (see `ApplyUnaryRpcDeadline()`).
* Finally, for all requests, it applies any `GrpcSetupOption`; this runs last
* so applications can override the deadline.
*
* @see https://cloud.google.com/apis/docs/system-parameters
*/
template <typename Request>
void ApplyQueryParameters(grpc::ClientContext& ctx, Options const& options,
Request const& request) {
using HasStandardParameters =
std::integral_constant<bool,
storage::internal::SupportsOption<
Request, storage::QuotaUser>::type::value &&
storage::internal::SupportsOption<
Request, storage::UserIp>::type::value &&
storage::internal::SupportsOption<
Request, storage::Fields>::type::value>;
ApplyQuotaUserAndFieldMask(ctx, request, HasStandardParameters{});
if (!IsStreamingRequest<Request>::value) {
ApplyUnaryRpcDeadline(ctx, options);
}
internal::ConfigureContext(ctx, options);
}

Expand Down
126 changes: 126 additions & 0 deletions google/cloud/storage/internal/grpc/configure_client_context_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
#include "google/cloud/storage/internal/grpc/configure_client_context.h"
#include "google/cloud/storage/internal/hash_function.h"
#include "google/cloud/storage/internal/object_requests.h"
#include "google/cloud/storage/internal/sign_blob_requests.h"
#include "google/cloud/grpc_options.h"
#include "google/cloud/testing_util/validate_metadata.h"
#include <gmock/gmock.h>
Expand Down Expand Up @@ -140,6 +141,131 @@ TEST_F(GrpcConfigureClientContext, ApplyQueryParametersGrpcOptions) {
storage::internal::ReadObjectRangeRequest("test-bucket", "test-object"));
}

TEST_F(GrpcConfigureClientContext, ApplyQueryParametersUnaryDeadline) {
std::chrono::seconds const timeout(42);
Options const options =
Options{}.set<storage::TransferStallTimeoutOption>(timeout);

std::chrono::system_clock::time_point const before =
std::chrono::system_clock::now();
grpc::ClientContext ctx;
ApplyQueryParameters(ctx, options,
storage::internal::GetObjectMetadataRequest(
"test-bucket", "test-object"));
std::chrono::system_clock::time_point const after =
std::chrono::system_clock::now();

EXPECT_GE(ctx.deadline(), before + timeout);
EXPECT_LE(ctx.deadline(), after + timeout);
}

TEST_F(GrpcConfigureClientContext, ApplyQueryParametersSignBlobUnaryDeadline) {
std::chrono::seconds const timeout(15);
Options const options =
Options{}.set<storage::TransferStallTimeoutOption>(timeout);

std::chrono::system_clock::time_point const before =
std::chrono::system_clock::now();
grpc::ClientContext ctx;
ApplyQueryParameters(
ctx, options,
storage::internal::SignBlobRequest("test-sa", "dGVzdA==", {}));
std::chrono::system_clock::time_point const after =
std::chrono::system_clock::now();

EXPECT_GE(ctx.deadline(), before + timeout);
EXPECT_LE(ctx.deadline(), after + timeout);
}

TEST_F(GrpcConfigureClientContext, ApplyQueryParametersUnaryDeadlineZero) {
Comment thread
ajayky-os marked this conversation as resolved.
Options const options = Options{}.set<storage::TransferStallTimeoutOption>(
std::chrono::seconds(0));

grpc::ClientContext ctx;
ApplyQueryParameters(ctx, options,
storage::internal::GetObjectMetadataRequest(
"test-bucket", "test-object"));
EXPECT_EQ(ctx.deadline(), std::chrono::system_clock::time_point::max());
}

TEST_F(GrpcConfigureClientContext, ApplyQueryParametersUnaryDeadlineNegative) {
Options const options = Options{}.set<storage::TransferStallTimeoutOption>(
std::chrono::seconds(-1));

grpc::ClientContext ctx;
ApplyQueryParameters(ctx, options,
storage::internal::GetObjectMetadataRequest(
"test-bucket", "test-object"));
EXPECT_EQ(ctx.deadline(), std::chrono::system_clock::time_point::max());
}

TEST_F(GrpcConfigureClientContext,
ApplyQueryParametersUnaryDeadlineRespectsExistingDeadline) {
std::chrono::system_clock::time_point const existing_deadline =
std::chrono::system_clock::now() + std::chrono::seconds(5);
Options const options = Options{}.set<storage::TransferStallTimeoutOption>(
std::chrono::seconds(120));

grpc::ClientContext ctx;
ctx.set_deadline(existing_deadline);
ApplyQueryParameters(ctx, options,
storage::internal::GetObjectMetadataRequest(
"test-bucket", "test-object"));
// gRPC converts system_clock::time_point to gpr_timespec (nanosecond
// resolution), allow 1ms tolerance for clock conversion.
EXPECT_LE(std::chrono::abs(ctx.deadline() - existing_deadline),
std::chrono::milliseconds(1));
}

TEST_F(GrpcConfigureClientContext,
ApplyQueryParametersUnaryDeadlineOverriddenByGrpcSetupOption) {
std::chrono::system_clock::time_point const custom_deadline =
std::chrono::system_clock::now() + std::chrono::seconds(7);
Options const options =
Options{}
.set<storage::TransferStallTimeoutOption>(std::chrono::seconds(120))
.set<google::cloud::internal::GrpcSetupOption>(
[custom_deadline](grpc::ClientContext& c) {
c.set_deadline(custom_deadline);
});

grpc::ClientContext ctx;
ApplyQueryParameters(ctx, options,
storage::internal::GetObjectMetadataRequest(
"test-bucket", "test-object"));
EXPECT_LE(std::chrono::abs(ctx.deadline() - custom_deadline),
std::chrono::milliseconds(1));
}

TEST_F(GrpcConfigureClientContext,
ApplyQueryParametersStreamingRequestsExcludeWholeRpcDeadline) {
Options const options = Options{}.set<storage::TransferStallTimeoutOption>(
std::chrono::seconds(42));

{
grpc::ClientContext ctx;
ApplyQueryParameters(ctx, options,
storage::internal::ReadObjectRangeRequest(
"test-bucket", "test-object"));
EXPECT_EQ(ctx.deadline(), std::chrono::system_clock::time_point::max());
}
{
grpc::ClientContext ctx;
ApplyQueryParameters(ctx, options,
storage::internal::InsertObjectMediaRequest(
"test-bucket", "test-object", "content"));
EXPECT_EQ(ctx.deadline(), std::chrono::system_clock::time_point::max());
}
{
grpc::ClientContext ctx;
ApplyQueryParameters(ctx, options,
storage::internal::UploadChunkRequest(
"projects/_/buckets/test-bucket/upload-id", 0, {},
CreateNullHashFunction()));
EXPECT_EQ(ctx.deadline(), std::chrono::system_clock::time_point::max());
}
}

TEST_F(GrpcConfigureClientContext, ApplyRoutingHeadersInsertObjectMedia) {
storage::internal::InsertObjectMediaRequest req("test-bucket", "test-object",
"content");
Expand Down
Loading
Loading