aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/core
diff options
context:
space:
mode:
authorGravatar Mark D. Roth <roth@google.com>2018-02-28 13:00:04 -0800
committerGravatar Mark D. Roth <roth@google.com>2018-02-28 13:00:04 -0800
commit718c8341ca8dcad17d93dc9e20c289c48b7ece34 (patch)
tree0a1b19d4c150373b860a5c236528030c150f4494 /test/core
parent0fc97adc9ee41d517ee49ec8e3a8338b793fba7e (diff)
Retry support.
Diffstat (limited to 'test/core')
-rw-r--r--test/core/client_channel/BUILD11
-rw-r--r--test/core/client_channel/status_util_test.cc49
-rw-r--r--test/core/end2end/end2end_nosec_tests.cc120
-rw-r--r--test/core/end2end/end2end_tests.cc120
-rw-r--r--test/core/end2end/fixtures/h2_proxy.cc12
-rw-r--r--test/core/end2end/fuzzers/hpack.dictionary8
-rwxr-xr-xtest/core/end2end/gen_build_yaml.py116
-rwxr-xr-xtest/core/end2end/generate_tests.bzl65
-rw-r--r--test/core/end2end/tests/retry.cc325
-rw-r--r--test/core/end2end/tests/retry_cancellation.cc277
-rw-r--r--test/core/end2end/tests/retry_disabled.cc262
-rw-r--r--test/core/end2end/tests/retry_exceeds_buffer_size_in_initial_batch.cc266
-rw-r--r--test/core/end2end/tests/retry_exceeds_buffer_size_in_subsequent_batch.cc276
-rw-r--r--test/core/end2end/tests/retry_non_retriable_status.cc257
-rw-r--r--test/core/end2end/tests/retry_recv_initial_metadata.cc268
-rw-r--r--test/core/end2end/tests/retry_recv_message.cc261
-rw-r--r--test/core/end2end/tests/retry_server_pushback_delay.cc318
-rw-r--r--test/core/end2end/tests/retry_server_pushback_disabled.cc306
-rw-r--r--test/core/end2end/tests/retry_streaming.cc424
-rw-r--r--test/core/end2end/tests/retry_streaming_after_commit.cc354
-rw-r--r--test/core/end2end/tests/retry_streaming_succeeds_before_replay_finished.cc400
-rw-r--r--test/core/end2end/tests/retry_throttled.cc264
-rw-r--r--test/core/end2end/tests/retry_too_many_attempts.cc299
-rw-r--r--test/core/transport/BUILD12
-rw-r--r--test/core/transport/status_metadata_test.cc61
25 files changed, 5090 insertions, 41 deletions
diff --git a/test/core/client_channel/BUILD b/test/core/client_channel/BUILD
index c4d12fe3a1..d430b722df 100644
--- a/test/core/client_channel/BUILD
+++ b/test/core/client_channel/BUILD
@@ -54,3 +54,14 @@ grpc_cc_test(
],
)
+grpc_cc_test(
+ name = "status_util_test",
+ srcs = ["status_util_test.cc"],
+ language = "C++",
+ deps = [
+ "//:grpc",
+ ],
+ external_deps = [
+ "gtest",
+ ],
+)
diff --git a/test/core/client_channel/status_util_test.cc b/test/core/client_channel/status_util_test.cc
new file mode 100644
index 0000000000..f944990ad2
--- /dev/null
+++ b/test/core/client_channel/status_util_test.cc
@@ -0,0 +1,49 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "src/core/ext/filters/client_channel/status_util.h"
+
+#include <gtest/gtest.h>
+
+namespace grpc_core {
+namespace internal {
+namespace {
+
+TEST(StatusCodeSet, Basic) {
+ StatusCodeSet set;
+ EXPECT_TRUE(set.Empty());
+ EXPECT_FALSE(set.Contains(GRPC_STATUS_OK));
+ EXPECT_FALSE(set.Contains(GRPC_STATUS_UNAVAILABLE));
+ set.Add(GRPC_STATUS_OK);
+ EXPECT_FALSE(set.Empty());
+ EXPECT_TRUE(set.Contains(GRPC_STATUS_OK));
+ EXPECT_FALSE(set.Contains(GRPC_STATUS_UNAVAILABLE));
+ set.Add(GRPC_STATUS_UNAVAILABLE);
+ EXPECT_FALSE(set.Empty());
+ EXPECT_TRUE(set.Contains(GRPC_STATUS_OK));
+ EXPECT_TRUE(set.Contains(GRPC_STATUS_UNAVAILABLE));
+}
+
+} // namespace
+} // namespace internal
+} // namespace grpc_core
+
+int main(int argc, char** argv) {
+ ::testing::InitGoogleTest(&argc, argv);
+ return RUN_ALL_TESTS();
+}
diff --git a/test/core/end2end/end2end_nosec_tests.cc b/test/core/end2end/end2end_nosec_tests.cc
index 6318550ad8..78ddcdb8f6 100644
--- a/test/core/end2end/end2end_nosec_tests.cc
+++ b/test/core/end2end/end2end_nosec_tests.cc
@@ -118,6 +118,36 @@ extern void request_with_payload(grpc_end2end_test_config config);
extern void request_with_payload_pre_init(void);
extern void resource_quota_server(grpc_end2end_test_config config);
extern void resource_quota_server_pre_init(void);
+extern void retry(grpc_end2end_test_config config);
+extern void retry_pre_init(void);
+extern void retry_cancellation(grpc_end2end_test_config config);
+extern void retry_cancellation_pre_init(void);
+extern void retry_disabled(grpc_end2end_test_config config);
+extern void retry_disabled_pre_init(void);
+extern void retry_exceeds_buffer_size_in_initial_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_initial_batch_pre_init(void);
+extern void retry_exceeds_buffer_size_in_subsequent_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_subsequent_batch_pre_init(void);
+extern void retry_non_retriable_status(grpc_end2end_test_config config);
+extern void retry_non_retriable_status_pre_init(void);
+extern void retry_recv_initial_metadata(grpc_end2end_test_config config);
+extern void retry_recv_initial_metadata_pre_init(void);
+extern void retry_recv_message(grpc_end2end_test_config config);
+extern void retry_recv_message_pre_init(void);
+extern void retry_server_pushback_delay(grpc_end2end_test_config config);
+extern void retry_server_pushback_delay_pre_init(void);
+extern void retry_server_pushback_disabled(grpc_end2end_test_config config);
+extern void retry_server_pushback_disabled_pre_init(void);
+extern void retry_streaming(grpc_end2end_test_config config);
+extern void retry_streaming_pre_init(void);
+extern void retry_streaming_after_commit(grpc_end2end_test_config config);
+extern void retry_streaming_after_commit_pre_init(void);
+extern void retry_streaming_succeeds_before_replay_finished(grpc_end2end_test_config config);
+extern void retry_streaming_succeeds_before_replay_finished_pre_init(void);
+extern void retry_throttled(grpc_end2end_test_config config);
+extern void retry_throttled_pre_init(void);
+extern void retry_too_many_attempts(grpc_end2end_test_config config);
+extern void retry_too_many_attempts_pre_init(void);
extern void server_finishes_request(grpc_end2end_test_config config);
extern void server_finishes_request_pre_init(void);
extern void shutdown_finishes_calls(grpc_end2end_test_config config);
@@ -197,6 +227,21 @@ void grpc_end2end_tests_pre_init(void) {
request_with_flags_pre_init();
request_with_payload_pre_init();
resource_quota_server_pre_init();
+ retry_pre_init();
+ retry_cancellation_pre_init();
+ retry_disabled_pre_init();
+ retry_exceeds_buffer_size_in_initial_batch_pre_init();
+ retry_exceeds_buffer_size_in_subsequent_batch_pre_init();
+ retry_non_retriable_status_pre_init();
+ retry_recv_initial_metadata_pre_init();
+ retry_recv_message_pre_init();
+ retry_server_pushback_delay_pre_init();
+ retry_server_pushback_disabled_pre_init();
+ retry_streaming_pre_init();
+ retry_streaming_after_commit_pre_init();
+ retry_streaming_succeeds_before_replay_finished_pre_init();
+ retry_throttled_pre_init();
+ retry_too_many_attempts_pre_init();
server_finishes_request_pre_init();
shutdown_finishes_calls_pre_init();
shutdown_finishes_tags_pre_init();
@@ -265,6 +310,21 @@ void grpc_end2end_tests(int argc, char **argv,
request_with_flags(config);
request_with_payload(config);
resource_quota_server(config);
+ retry(config);
+ retry_cancellation(config);
+ retry_disabled(config);
+ retry_exceeds_buffer_size_in_initial_batch(config);
+ retry_exceeds_buffer_size_in_subsequent_batch(config);
+ retry_non_retriable_status(config);
+ retry_recv_initial_metadata(config);
+ retry_recv_message(config);
+ retry_server_pushback_delay(config);
+ retry_server_pushback_disabled(config);
+ retry_streaming(config);
+ retry_streaming_after_commit(config);
+ retry_streaming_succeeds_before_replay_finished(config);
+ retry_throttled(config);
+ retry_too_many_attempts(config);
server_finishes_request(config);
shutdown_finishes_calls(config);
shutdown_finishes_tags(config);
@@ -460,6 +520,66 @@ void grpc_end2end_tests(int argc, char **argv,
resource_quota_server(config);
continue;
}
+ if (0 == strcmp("retry", argv[i])) {
+ retry(config);
+ continue;
+ }
+ if (0 == strcmp("retry_cancellation", argv[i])) {
+ retry_cancellation(config);
+ continue;
+ }
+ if (0 == strcmp("retry_disabled", argv[i])) {
+ retry_disabled(config);
+ continue;
+ }
+ if (0 == strcmp("retry_exceeds_buffer_size_in_initial_batch", argv[i])) {
+ retry_exceeds_buffer_size_in_initial_batch(config);
+ continue;
+ }
+ if (0 == strcmp("retry_exceeds_buffer_size_in_subsequent_batch", argv[i])) {
+ retry_exceeds_buffer_size_in_subsequent_batch(config);
+ continue;
+ }
+ if (0 == strcmp("retry_non_retriable_status", argv[i])) {
+ retry_non_retriable_status(config);
+ continue;
+ }
+ if (0 == strcmp("retry_recv_initial_metadata", argv[i])) {
+ retry_recv_initial_metadata(config);
+ continue;
+ }
+ if (0 == strcmp("retry_recv_message", argv[i])) {
+ retry_recv_message(config);
+ continue;
+ }
+ if (0 == strcmp("retry_server_pushback_delay", argv[i])) {
+ retry_server_pushback_delay(config);
+ continue;
+ }
+ if (0 == strcmp("retry_server_pushback_disabled", argv[i])) {
+ retry_server_pushback_disabled(config);
+ continue;
+ }
+ if (0 == strcmp("retry_streaming", argv[i])) {
+ retry_streaming(config);
+ continue;
+ }
+ if (0 == strcmp("retry_streaming_after_commit", argv[i])) {
+ retry_streaming_after_commit(config);
+ continue;
+ }
+ if (0 == strcmp("retry_streaming_succeeds_before_replay_finished", argv[i])) {
+ retry_streaming_succeeds_before_replay_finished(config);
+ continue;
+ }
+ if (0 == strcmp("retry_throttled", argv[i])) {
+ retry_throttled(config);
+ continue;
+ }
+ if (0 == strcmp("retry_too_many_attempts", argv[i])) {
+ retry_too_many_attempts(config);
+ continue;
+ }
if (0 == strcmp("server_finishes_request", argv[i])) {
server_finishes_request(config);
continue;
diff --git a/test/core/end2end/end2end_tests.cc b/test/core/end2end/end2end_tests.cc
index 9d8dfd6723..fb1e61b39f 100644
--- a/test/core/end2end/end2end_tests.cc
+++ b/test/core/end2end/end2end_tests.cc
@@ -120,6 +120,36 @@ extern void request_with_payload(grpc_end2end_test_config config);
extern void request_with_payload_pre_init(void);
extern void resource_quota_server(grpc_end2end_test_config config);
extern void resource_quota_server_pre_init(void);
+extern void retry(grpc_end2end_test_config config);
+extern void retry_pre_init(void);
+extern void retry_cancellation(grpc_end2end_test_config config);
+extern void retry_cancellation_pre_init(void);
+extern void retry_disabled(grpc_end2end_test_config config);
+extern void retry_disabled_pre_init(void);
+extern void retry_exceeds_buffer_size_in_initial_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_initial_batch_pre_init(void);
+extern void retry_exceeds_buffer_size_in_subsequent_batch(grpc_end2end_test_config config);
+extern void retry_exceeds_buffer_size_in_subsequent_batch_pre_init(void);
+extern void retry_non_retriable_status(grpc_end2end_test_config config);
+extern void retry_non_retriable_status_pre_init(void);
+extern void retry_recv_initial_metadata(grpc_end2end_test_config config);
+extern void retry_recv_initial_metadata_pre_init(void);
+extern void retry_recv_message(grpc_end2end_test_config config);
+extern void retry_recv_message_pre_init(void);
+extern void retry_server_pushback_delay(grpc_end2end_test_config config);
+extern void retry_server_pushback_delay_pre_init(void);
+extern void retry_server_pushback_disabled(grpc_end2end_test_config config);
+extern void retry_server_pushback_disabled_pre_init(void);
+extern void retry_streaming(grpc_end2end_test_config config);
+extern void retry_streaming_pre_init(void);
+extern void retry_streaming_after_commit(grpc_end2end_test_config config);
+extern void retry_streaming_after_commit_pre_init(void);
+extern void retry_streaming_succeeds_before_replay_finished(grpc_end2end_test_config config);
+extern void retry_streaming_succeeds_before_replay_finished_pre_init(void);
+extern void retry_throttled(grpc_end2end_test_config config);
+extern void retry_throttled_pre_init(void);
+extern void retry_too_many_attempts(grpc_end2end_test_config config);
+extern void retry_too_many_attempts_pre_init(void);
extern void server_finishes_request(grpc_end2end_test_config config);
extern void server_finishes_request_pre_init(void);
extern void shutdown_finishes_calls(grpc_end2end_test_config config);
@@ -200,6 +230,21 @@ void grpc_end2end_tests_pre_init(void) {
request_with_flags_pre_init();
request_with_payload_pre_init();
resource_quota_server_pre_init();
+ retry_pre_init();
+ retry_cancellation_pre_init();
+ retry_disabled_pre_init();
+ retry_exceeds_buffer_size_in_initial_batch_pre_init();
+ retry_exceeds_buffer_size_in_subsequent_batch_pre_init();
+ retry_non_retriable_status_pre_init();
+ retry_recv_initial_metadata_pre_init();
+ retry_recv_message_pre_init();
+ retry_server_pushback_delay_pre_init();
+ retry_server_pushback_disabled_pre_init();
+ retry_streaming_pre_init();
+ retry_streaming_after_commit_pre_init();
+ retry_streaming_succeeds_before_replay_finished_pre_init();
+ retry_throttled_pre_init();
+ retry_too_many_attempts_pre_init();
server_finishes_request_pre_init();
shutdown_finishes_calls_pre_init();
shutdown_finishes_tags_pre_init();
@@ -269,6 +314,21 @@ void grpc_end2end_tests(int argc, char **argv,
request_with_flags(config);
request_with_payload(config);
resource_quota_server(config);
+ retry(config);
+ retry_cancellation(config);
+ retry_disabled(config);
+ retry_exceeds_buffer_size_in_initial_batch(config);
+ retry_exceeds_buffer_size_in_subsequent_batch(config);
+ retry_non_retriable_status(config);
+ retry_recv_initial_metadata(config);
+ retry_recv_message(config);
+ retry_server_pushback_delay(config);
+ retry_server_pushback_disabled(config);
+ retry_streaming(config);
+ retry_streaming_after_commit(config);
+ retry_streaming_succeeds_before_replay_finished(config);
+ retry_throttled(config);
+ retry_too_many_attempts(config);
server_finishes_request(config);
shutdown_finishes_calls(config);
shutdown_finishes_tags(config);
@@ -468,6 +528,66 @@ void grpc_end2end_tests(int argc, char **argv,
resource_quota_server(config);
continue;
}
+ if (0 == strcmp("retry", argv[i])) {
+ retry(config);
+ continue;
+ }
+ if (0 == strcmp("retry_cancellation", argv[i])) {
+ retry_cancellation(config);
+ continue;
+ }
+ if (0 == strcmp("retry_disabled", argv[i])) {
+ retry_disabled(config);
+ continue;
+ }
+ if (0 == strcmp("retry_exceeds_buffer_size_in_initial_batch", argv[i])) {
+ retry_exceeds_buffer_size_in_initial_batch(config);
+ continue;
+ }
+ if (0 == strcmp("retry_exceeds_buffer_size_in_subsequent_batch", argv[i])) {
+ retry_exceeds_buffer_size_in_subsequent_batch(config);
+ continue;
+ }
+ if (0 == strcmp("retry_non_retriable_status", argv[i])) {
+ retry_non_retriable_status(config);
+ continue;
+ }
+ if (0 == strcmp("retry_recv_initial_metadata", argv[i])) {
+ retry_recv_initial_metadata(config);
+ continue;
+ }
+ if (0 == strcmp("retry_recv_message", argv[i])) {
+ retry_recv_message(config);
+ continue;
+ }
+ if (0 == strcmp("retry_server_pushback_delay", argv[i])) {
+ retry_server_pushback_delay(config);
+ continue;
+ }
+ if (0 == strcmp("retry_server_pushback_disabled", argv[i])) {
+ retry_server_pushback_disabled(config);
+ continue;
+ }
+ if (0 == strcmp("retry_streaming", argv[i])) {
+ retry_streaming(config);
+ continue;
+ }
+ if (0 == strcmp("retry_streaming_after_commit", argv[i])) {
+ retry_streaming_after_commit(config);
+ continue;
+ }
+ if (0 == strcmp("retry_streaming_succeeds_before_replay_finished", argv[i])) {
+ retry_streaming_succeeds_before_replay_finished(config);
+ continue;
+ }
+ if (0 == strcmp("retry_throttled", argv[i])) {
+ retry_throttled(config);
+ continue;
+ }
+ if (0 == strcmp("retry_too_many_attempts", argv[i])) {
+ retry_too_many_attempts(config);
+ continue;
+ }
if (0 == strcmp("server_finishes_request", argv[i])) {
server_finishes_request(config);
continue;
diff --git a/test/core/end2end/fixtures/h2_proxy.cc b/test/core/end2end/fixtures/h2_proxy.cc
index 93dde4f4a2..28a6eeeb78 100644
--- a/test/core/end2end/fixtures/h2_proxy.cc
+++ b/test/core/end2end/fixtures/h2_proxy.cc
@@ -49,7 +49,17 @@ static grpc_server* create_proxy_server(const char* port,
static grpc_channel* create_proxy_client(const char* target,
grpc_channel_args* client_args) {
- return grpc_insecure_channel_create(target, client_args, nullptr);
+ // Disable retries in proxy client.
+ grpc_arg arg;
+ arg.type = GRPC_ARG_INTEGER;
+ arg.key = const_cast<char*>(GRPC_ARG_ENABLE_RETRIES);
+ arg.value.integer = 0;
+ grpc_channel_args* new_args =
+ grpc_channel_args_copy_and_add(client_args, &arg, 1);
+ grpc_channel* channel =
+ grpc_insecure_channel_create(target, new_args, nullptr);
+ grpc_channel_args_destroy(new_args);
+ return channel;
}
static const grpc_end2end_proxy_def proxy_def = {create_proxy_server,
diff --git a/test/core/end2end/fuzzers/hpack.dictionary b/test/core/end2end/fuzzers/hpack.dictionary
index 3ed82e19bd..c719d0fa84 100644
--- a/test/core/end2end/fuzzers/hpack.dictionary
+++ b/test/core/end2end/fuzzers/hpack.dictionary
@@ -21,7 +21,13 @@
"\x0Auser-agent"
"\x04host"
"\x08lb-token"
+"\x1Agrpc-previous-rpc-attempts"
+"\x16grpc-retry-pushback-ms"
"\x0Cgrpc-timeout"
+"\x011"
+"\x012"
+"\x013"
+"\x014"
"\x00"
"\x13grpc.wait_for_ready"
"\x0Cgrpc.timeout"
@@ -32,8 +38,6 @@
"\x04gzip"
"\x0Bstream/gzip"
"\x010"
-"\x011"
-"\x012"
"\x08identity"
"\x08trailers"
"\x10application/grpc"
diff --git a/test/core/end2end/gen_build_yaml.py b/test/core/end2end/gen_build_yaml.py
index e7cf97b2d0..4e20b0b334 100755
--- a/test/core/end2end/gen_build_yaml.py
+++ b/test/core/end2end/gen_build_yaml.py
@@ -24,15 +24,24 @@ import hashlib
FixtureOptions = collections.namedtuple(
'FixtureOptions',
- 'fullstack includes_proxy dns_resolver name_resolution secure platforms ci_mac tracing exclude_configs exclude_iomgrs large_writes enables_compression supports_compression is_inproc is_http2 supports_proxy_auth supports_write_buffering')
+ 'fullstack includes_proxy dns_resolver name_resolution secure platforms ci_mac tracing exclude_configs exclude_iomgrs large_writes enables_compression supports_compression is_inproc is_http2 supports_proxy_auth supports_write_buffering client_channel')
default_unsecure_fixture_options = FixtureOptions(
- True, False, True, True, False, ['windows', 'linux', 'mac', 'posix'], True, False, [], [], True, False, True, False, True, False, True)
-socketpair_unsecure_fixture_options = default_unsecure_fixture_options._replace(fullstack=False, dns_resolver=False)
-default_secure_fixture_options = default_unsecure_fixture_options._replace(secure=True)
-uds_fixture_options = default_unsecure_fixture_options._replace(dns_resolver=False, platforms=['linux', 'mac', 'posix'], exclude_iomgrs=['uv'])
+ True, False, True, True, False, ['windows', 'linux', 'mac', 'posix'],
+ True, False, [], [], True, False, True, False, True, False, True, True)
+socketpair_unsecure_fixture_options = default_unsecure_fixture_options._replace(
+ fullstack=False, dns_resolver=False, client_channel=False)
+default_secure_fixture_options = default_unsecure_fixture_options._replace(
+ secure=True)
+uds_fixture_options = default_unsecure_fixture_options._replace(
+ dns_resolver=False, platforms=['linux', 'mac', 'posix'],
+ exclude_iomgrs=['uv'])
fd_unsecure_fixture_options = default_unsecure_fixture_options._replace(
- dns_resolver=False, fullstack=False, platforms=['linux', 'mac', 'posix'], exclude_iomgrs=['uv'])
-inproc_fixture_options = default_unsecure_fixture_options._replace(dns_resolver=False, fullstack=False, name_resolution=False, supports_compression=False, is_inproc=True, is_http2=False, supports_write_buffering=False)
+ dns_resolver=False, fullstack=False, platforms=['linux', 'mac', 'posix'],
+ exclude_iomgrs=['uv'], client_channel=False)
+inproc_fixture_options = default_unsecure_fixture_options._replace(
+ dns_resolver=False, fullstack=False, name_resolution=False,
+ supports_compression=False, is_inproc=True, is_http2=False,
+ supports_write_buffering=False, client_channel=False)
# maps fixture name to whether it requires the security library
END2END_FIXTURES = {
@@ -68,9 +77,12 @@ END2END_FIXTURES = {
TestOptions = collections.namedtuple(
'TestOptions',
- 'needs_fullstack needs_dns needs_names proxyable secure traceable cpu_cost exclude_iomgrs large_writes flaky allows_compression needs_compression exclude_inproc needs_http2 needs_proxy_auth needs_write_buffering')
-default_test_options = TestOptions(False, False, False, True, False, True, 1.0, [], False, False, True, False, False, False, False, False)
-connectivity_test_options = default_test_options._replace(needs_fullstack=True)
+ 'needs_fullstack needs_dns needs_names proxyable secure traceable cpu_cost exclude_iomgrs large_writes flaky allows_compression needs_compression exclude_inproc needs_http2 needs_proxy_auth needs_write_buffering needs_client_channel')
+default_test_options = TestOptions(
+ False, False, False, True, False, True, 1.0, [], False, False, True,
+ False, False, False, False, False, False)
+connectivity_test_options = default_test_options._replace(
+ needs_fullstack=True)
LOWCPU = 0.1
@@ -80,9 +92,8 @@ END2END_TESTS = {
'bad_hostname': default_test_options._replace(needs_names=True),
'bad_ping': connectivity_test_options._replace(proxyable=False),
'binary_metadata': default_test_options._replace(cpu_cost=LOWCPU),
- 'resource_quota_server': default_test_options._replace(large_writes=True,
- proxyable=False,
- allows_compression=False),
+ 'resource_quota_server': default_test_options._replace(
+ large_writes=True, proxyable=False, allows_compression=False),
'call_creds': default_test_options._replace(secure=True),
'cancel_after_accept': default_test_options._replace(cpu_cost=LOWCPU),
'cancel_after_client_done': default_test_options._replace(cpu_cost=LOWCPU),
@@ -91,18 +102,21 @@ END2END_TESTS = {
'cancel_before_invoke': default_test_options._replace(cpu_cost=LOWCPU),
'cancel_in_a_vacuum': default_test_options._replace(cpu_cost=LOWCPU),
'cancel_with_status': default_test_options._replace(cpu_cost=LOWCPU),
- 'compressed_payload': default_test_options._replace(proxyable=False,needs_compression=True),
+ 'compressed_payload': default_test_options._replace(proxyable=False,
+ needs_compression=True),
'connectivity': connectivity_test_options._replace(needs_names=True,
proxyable=False, cpu_cost=LOWCPU, exclude_iomgrs=['uv']),
- 'default_host': default_test_options._replace(needs_fullstack=True,
- needs_dns=True,needs_names=True),
- 'disappearing_server': connectivity_test_options._replace(flaky=True,needs_names=True),
+ 'default_host': default_test_options._replace(
+ needs_fullstack=True, needs_dns=True, needs_names=True),
+ 'disappearing_server': connectivity_test_options._replace(flaky=True,
+ needs_names=True),
'empty_batch': default_test_options._replace(cpu_cost=LOWCPU),
'filter_causes_close': default_test_options._replace(cpu_cost=LOWCPU),
'filter_call_init_fails': default_test_options,
'filter_latency': default_test_options._replace(cpu_cost=LOWCPU),
'filter_status_code': default_test_options._replace(cpu_cost=LOWCPU),
- 'graceful_server_shutdown': default_test_options._replace(cpu_cost=LOWCPU,exclude_inproc=True),
+ 'graceful_server_shutdown': default_test_options._replace(
+ cpu_cost=LOWCPU, exclude_inproc=True),
'hpack_size': default_test_options._replace(proxyable=False,
traceable=False,
cpu_cost=LOWCPU),
@@ -127,30 +141,75 @@ END2END_TESTS = {
'payload': default_test_options,
'load_reporting_hook': default_test_options,
'ping_pong_streaming': default_test_options._replace(cpu_cost=LOWCPU),
- 'ping': connectivity_test_options._replace(proxyable=False, cpu_cost=LOWCPU),
+ 'ping': connectivity_test_options._replace(proxyable=False,
+ cpu_cost=LOWCPU),
'proxy_auth': default_test_options._replace(needs_proxy_auth=True),
'registered_call': default_test_options,
'request_with_flags': default_test_options._replace(
proxyable=False, cpu_cost=LOWCPU),
'request_with_payload': default_test_options._replace(cpu_cost=LOWCPU),
+ # TODO(roth): Remove proxyable=False for all retry tests once we
+ # have a way for the proxy to propagate the fact that trailing
+ # metadata is available when initial metadata is returned.
+ # See https://github.com/grpc/grpc/issues/14467 for context.
+ 'retry': default_test_options._replace(cpu_cost=LOWCPU,
+ needs_client_channel=True,
+ proxyable=False),
+ 'retry_cancellation': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_disabled': default_test_options._replace(cpu_cost=LOWCPU,
+ needs_client_channel=True,
+ proxyable=False),
+ 'retry_exceeds_buffer_size_in_initial_batch': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_exceeds_buffer_size_in_subsequent_batch':
+ default_test_options._replace(cpu_cost=LOWCPU,
+ needs_client_channel=True,
+ proxyable=False),
+ 'retry_non_retriable_status': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_recv_initial_metadata': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_recv_message': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_server_pushback_delay': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_server_pushback_disabled': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_streaming': default_test_options._replace(cpu_cost=LOWCPU,
+ needs_client_channel=True,
+ proxyable=False),
+ 'retry_streaming_after_commit': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
+ 'retry_streaming_succeeds_before_replay_finished':
+ default_test_options._replace(cpu_cost=LOWCPU,
+ needs_client_channel=True,
+ proxyable=False),
+ 'retry_throttled': default_test_options._replace(cpu_cost=LOWCPU,
+ needs_client_channel=True,
+ proxyable=False),
+ 'retry_too_many_attempts': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_client_channel=True, proxyable=False),
'server_finishes_request': default_test_options._replace(cpu_cost=LOWCPU),
'shutdown_finishes_calls': default_test_options._replace(cpu_cost=LOWCPU),
'shutdown_finishes_tags': default_test_options._replace(cpu_cost=LOWCPU),
'simple_cacheable_request': default_test_options._replace(cpu_cost=LOWCPU),
- 'stream_compression_compressed_payload': default_test_options._replace(proxyable=False,
- exclude_inproc=True),
- 'stream_compression_payload': default_test_options._replace(exclude_inproc=True),
- 'stream_compression_ping_pong_streaming': default_test_options._replace(exclude_inproc=True),
+ 'stream_compression_compressed_payload': default_test_options._replace(
+ proxyable=False, exclude_inproc=True),
+ 'stream_compression_payload': default_test_options._replace(
+ exclude_inproc=True),
+ 'stream_compression_ping_pong_streaming': default_test_options._replace(
+ exclude_inproc=True),
'simple_delayed_request': connectivity_test_options,
'simple_metadata': default_test_options,
'simple_request': default_test_options,
'streaming_error_response': default_test_options._replace(cpu_cost=LOWCPU),
'trailing_metadata': default_test_options,
'workaround_cronet_compression': default_test_options,
- 'write_buffering': default_test_options._replace(cpu_cost=LOWCPU,
- needs_write_buffering=True),
- 'write_buffering_at_end': default_test_options._replace(cpu_cost=LOWCPU,
- needs_write_buffering=True),
+ 'write_buffering': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_write_buffering=True),
+ 'write_buffering_at_end': default_test_options._replace(
+ cpu_cost=LOWCPU, needs_write_buffering=True),
}
@@ -191,6 +250,9 @@ def compatible(f, t):
if END2END_TESTS[t].needs_write_buffering:
if not END2END_FIXTURES[f].supports_write_buffering:
return False
+ if END2END_TESTS[t].needs_client_channel:
+ if not END2END_FIXTURES[f].client_channel:
+ return False
return True
diff --git a/test/core/end2end/generate_tests.bzl b/test/core/end2end/generate_tests.bzl
index bfd5d8525e..8e723fd609 100755
--- a/test/core/end2end/generate_tests.bzl
+++ b/test/core/end2end/generate_tests.bzl
@@ -24,7 +24,7 @@ def fixture_options(fullstack=True, includes_proxy=False, dns_resolver=True,
name_resolution=True, secure=True, tracing=False,
platforms=['windows', 'linux', 'mac', 'posix'],
is_inproc=False, is_http2=True, supports_proxy_auth=False,
- supports_write_buffering=True):
+ supports_write_buffering=True, client_channel=True):
return struct(
fullstack=fullstack,
includes_proxy=includes_proxy,
@@ -35,8 +35,9 @@ def fixture_options(fullstack=True, includes_proxy=False, dns_resolver=True,
is_inproc=is_inproc,
is_http2=is_http2,
supports_proxy_auth=supports_proxy_auth,
- supports_write_buffering=supports_write_buffering
- #platforms=platforms
+ supports_write_buffering=supports_write_buffering,
+ client_channel=client_channel,
+ #platforms=platforms,
)
@@ -47,6 +48,7 @@ END2END_FIXTURES = {
'h2_load_reporting': fixture_options(),
'h2_fakesec': fixture_options(),
'h2_fd': fixture_options(dns_resolver=False, fullstack=False,
+ client_channel=False,
platforms=['linux', 'mac', 'posix']),
'h2_full': fixture_options(),
'h2_full+pipe': fixture_options(platforms=['linux']),
@@ -55,24 +57,28 @@ END2END_FIXTURES = {
'h2_http_proxy': fixture_options(supports_proxy_auth=True),
'h2_oauth2': fixture_options(),
'h2_proxy': fixture_options(includes_proxy=True),
- 'h2_sockpair_1byte': fixture_options(fullstack=False, dns_resolver=False),
- 'h2_sockpair': fixture_options(fullstack=False, dns_resolver=False),
+ 'h2_sockpair_1byte': fixture_options(fullstack=False, dns_resolver=False,
+ client_channel=False),
+ 'h2_sockpair': fixture_options(fullstack=False, dns_resolver=False,
+ client_channel=False),
'h2_sockpair+trace': fixture_options(fullstack=False, dns_resolver=False,
- tracing=True),
+ tracing=True, client_channel=False),
'h2_ssl': fixture_options(secure=True),
'h2_ssl_proxy': fixture_options(includes_proxy=True, secure=True),
'h2_uds': fixture_options(dns_resolver=False,
platforms=['linux', 'mac', 'posix']),
'inproc': fixture_options(fullstack=False, dns_resolver=False,
name_resolution=False, is_inproc=True,
- is_http2=False, supports_write_buffering=False),
+ is_http2=False, supports_write_buffering=False,
+ client_channel=False),
}
def test_options(needs_fullstack=False, needs_dns=False, needs_names=False,
proxyable=True, secure=False, traceable=False,
exclude_inproc=False, needs_http2=False,
- needs_proxy_auth=False, needs_write_buffering=False):
+ needs_proxy_auth=False, needs_write_buffering=False,
+ needs_client_channel=False):
return struct(
needs_fullstack=needs_fullstack,
needs_dns=needs_dns,
@@ -83,7 +89,8 @@ def test_options(needs_fullstack=False, needs_dns=False, needs_names=False,
exclude_inproc=exclude_inproc,
needs_http2=needs_http2,
needs_proxy_auth=needs_proxy_auth,
- needs_write_buffering=needs_write_buffering
+ needs_write_buffering=needs_write_buffering,
+ needs_client_channel=needs_client_channel,
)
@@ -118,7 +125,8 @@ END2END_TESTS = {
'invoke_large_request': test_options(),
'keepalive_timeout': test_options(proxyable=False, needs_http2=True),
'large_metadata': test_options(),
- 'max_concurrent_streams': test_options(proxyable=False, exclude_inproc=True),
+ 'max_concurrent_streams': test_options(proxyable=False,
+ exclude_inproc=True),
'max_connection_age': test_options(exclude_inproc=True),
'max_connection_idle': test_options(needs_fullstack=True, proxyable=False),
'max_message_length': test_options(),
@@ -134,6 +142,37 @@ END2END_TESTS = {
'registered_call': test_options(),
'request_with_flags': test_options(proxyable=False),
'request_with_payload': test_options(),
+ # TODO(roth): Remove proxyable=False for all retry tests once we
+ # have a way for the proxy to propagate the fact that trailing
+ # metadata is available when initial metadata is returned.
+ # See https://github.com/grpc/grpc/issues/14467 for context.
+ 'retry': test_options(needs_client_channel=True, proxyable=False),
+ 'retry_cancellation': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_disabled': test_options(needs_client_channel=True, proxyable=False),
+ 'retry_exceeds_buffer_size_in_initial_batch': test_options(
+ needs_client_channel=True, proxyable=False),
+ 'retry_exceeds_buffer_size_in_subsequent_batch': test_options(
+ needs_client_channel=True, proxyable=False),
+ 'retry_non_retriable_status': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_recv_initial_metadata': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_recv_message': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_server_pushback_delay': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_server_pushback_disabled': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_streaming': test_options(needs_client_channel=True, proxyable=False),
+ 'retry_streaming_after_commit': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_streaming_succeeds_before_replay_finished': test_options(
+ needs_client_channel=True, proxyable=False),
+ 'retry_throttled': test_options(needs_client_channel=True,
+ proxyable=False),
+ 'retry_too_many_attempts': test_options(needs_client_channel=True,
+ proxyable=False),
'server_finishes_request': test_options(),
'shutdown_finishes_calls': test_options(),
'shutdown_finishes_tags': test_options(),
@@ -142,7 +181,8 @@ END2END_TESTS = {
'simple_metadata': test_options(),
'simple_request': test_options(),
'streaming_error_response': test_options(),
- 'stream_compression_compressed_payload': test_options(proxyable=False, exclude_inproc=True),
+ 'stream_compression_compressed_payload': test_options(proxyable=False,
+ exclude_inproc=True),
'stream_compression_payload': test_options(exclude_inproc=True),
'stream_compression_ping_pong_streaming': test_options(exclude_inproc=True),
'trailing_metadata': test_options(),
@@ -183,6 +223,9 @@ def compatible(fopt, topt):
if topt.needs_write_buffering:
if not fopt.supports_write_buffering:
return False
+ if topt.needs_client_channel:
+ if not fopt.client_channel:
+ return False
return True
diff --git a/test/core/end2end/tests/retry.cc b/test/core/end2end/tests/retry.cc
new file mode 100644
index 0000000000..38ecc6fbb1
--- /dev/null
+++ b/test/core/end2end/tests/retry.cc
@@ -0,0 +1,325 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests a basic retry scenario:
+// - 2 retries allowed for ABORTED status
+// - first attempt returns ABORTED
+// - second attempt returns OK
+static void test_retry(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 3,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ // Make sure the "grpc-previous-rpc-attempts" header was not sent in the
+ // initial attempt.
+ for (size_t i = 0; i < request_metadata_recv.count; ++i) {
+ GPR_ASSERT(!grpc_slice_eq(request_metadata_recv.metadata[i].key,
+ GRPC_MDSTR_GRPC_PREVIOUS_RPC_ATTEMPTS));
+ }
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ grpc_call_unref(s);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_call_details_init(&call_details);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(201));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+ cq_verify(cqv);
+
+ // Make sure the "grpc-previous-rpc-attempts" header was sent in the retry.
+ bool found_retry_header = false;
+ for (size_t i = 0; i < request_metadata_recv.count; ++i) {
+ if (grpc_slice_eq(request_metadata_recv.metadata[i].key,
+ GRPC_MDSTR_GRPC_PREVIOUS_RPC_ATTEMPTS)) {
+ GPR_ASSERT(
+ grpc_slice_eq(request_metadata_recv.metadata[i].value, GRPC_MDSTR_1));
+ found_retry_header = true;
+ break;
+ }
+ }
+ GPR_ASSERT(found_retry_header);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = response_payload;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_OK;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_OK);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 0);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry(config);
+}
+
+void retry_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_cancellation.cc b/test/core/end2end/tests/retry_cancellation.cc
new file mode 100644
index 0000000000..0504092c4f
--- /dev/null
+++ b/test/core/end2end/tests/retry_cancellation.cc
@@ -0,0 +1,277 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests retry cancellation.
+static void test_retry_cancellation(grpc_end2end_test_config config,
+ cancellation_mode mode) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 3,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " },\n"
+ " \"timeout\": \"5s\"\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ char* name;
+ gpr_asprintf(&name, "retry_cancellation/%s", mode.name);
+ grpc_end2end_test_fixture f = begin_test(config, name, &client_args, nullptr);
+ gpr_free(name);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ // Client starts a batch with all 6 ops.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ // Server gets a call and fails with retryable status.
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ grpc_call_unref(s);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_call_details_init(&call_details);
+
+ // Server gets a second call (the retry).
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(201));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+ cq_verify(cqv);
+
+ // Initiate cancellation.
+ GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c, nullptr));
+
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == mode.expect_status);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_cancellation(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ for (size_t i = 0; i < GPR_ARRAY_SIZE(cancellation_modes); ++i) {
+ test_retry_cancellation(config, cancellation_modes[i]);
+ }
+}
+
+void retry_cancellation_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_disabled.cc b/test/core/end2end/tests/retry_disabled.cc
new file mode 100644
index 0000000000..cb30502aeb
--- /dev/null
+++ b/test/core/end2end/tests/retry_disabled.cc
@@ -0,0 +1,262 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry when retries are disabled via the
+// GRPC_ARG_ENABLE_RETRIES channel arg, even when there is retry
+// configuration in the service config.
+// - 1 retry allowed for ABORTED status
+// - first attempt returns ABORTED but does not retry
+static void test_retry_disabled(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg args[2];
+ args[0].type = GRPC_ARG_STRING;
+ args[0].key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ args[0].value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ args[1].type = GRPC_ARG_INTEGER;
+ args[1].key = const_cast<char*>(GRPC_ARG_ENABLE_RETRIES);
+ args[1].value.integer = 0;
+ grpc_channel_args client_args = {GPR_ARRAY_SIZE(args), args};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_disabled", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_disabled(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_disabled(config);
+}
+
+void retry_disabled_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_exceeds_buffer_size_in_initial_batch.cc b/test/core/end2end/tests/retry_exceeds_buffer_size_in_initial_batch.cc
new file mode 100644
index 0000000000..3908f29971
--- /dev/null
+++ b/test/core/end2end/tests/retry_exceeds_buffer_size_in_initial_batch.cc
@@ -0,0 +1,266 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't make any further attempts after we exceed the
+// max buffer size.
+// - 1 retry allowed for ABORTED status
+// - buffer size set to 2 bytes
+// - client sends a 3-byte message
+// - first attempt gets ABORTED but is not retried
+static void test_retry_exceeds_buffer_size_in_initial_batch(
+ grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg args[2];
+ args[0].type = GRPC_ARG_STRING;
+ args[0].key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ args[0].value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ args[1].type = GRPC_ARG_INTEGER;
+ args[1].key = const_cast<char*>(GRPC_ARG_PER_RPC_RETRY_BUFFER_SIZE);
+ args[1].value.integer = 2;
+ grpc_channel_args client_args = {GPR_ARRAY_SIZE(args), args};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_exceeds_buffer_size_in_initial_batch",
+ &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_exceeds_buffer_size_in_initial_batch(
+ grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_exceeds_buffer_size_in_initial_batch(config);
+}
+
+void retry_exceeds_buffer_size_in_initial_batch_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_exceeds_buffer_size_in_subsequent_batch.cc b/test/core/end2end/tests/retry_exceeds_buffer_size_in_subsequent_batch.cc
new file mode 100644
index 0000000000..a8d7f0851a
--- /dev/null
+++ b/test/core/end2end/tests/retry_exceeds_buffer_size_in_subsequent_batch.cc
@@ -0,0 +1,276 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Similar to the retry_exceeds_buffer_size_in_initial_batch test, but we
+// don't exceed the buffer size until the second batch.
+// - 1 retry allowed for ABORTED status
+// - buffer size set to 100 KiB (larger than initial metadata)
+// - client sends a 100 KiB message
+// - first attempt gets ABORTED but is not retried
+static void test_retry_exceeds_buffer_size_in_subsequent_batch(
+ grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ char buf[102401];
+ memset(buf, 'a', sizeof(buf) - 1);
+ buf[sizeof(buf) - 1] = '\0';
+ grpc_slice request_payload_slice = grpc_slice_from_static_string(buf);
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg args[2];
+ args[0].type = GRPC_ARG_STRING;
+ args[0].key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ args[0].value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ args[1].type = GRPC_ARG_INTEGER;
+ args[1].key = const_cast<char*>(GRPC_ARG_PER_RPC_RETRY_BUFFER_SIZE);
+ args[1].value.integer = 102400;
+ grpc_channel_args client_args = {GPR_ARRAY_SIZE(args), args};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_exceeds_buffer_size_in_subsequent_batch",
+ &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_exceeds_buffer_size_in_subsequent_batch(
+ grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_exceeds_buffer_size_in_subsequent_batch(config);
+}
+
+void retry_exceeds_buffer_size_in_subsequent_batch_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_non_retriable_status.cc b/test/core/end2end/tests/retry_non_retriable_status.cc
new file mode 100644
index 0000000000..6d59db0b04
--- /dev/null
+++ b/test/core/end2end/tests/retry_non_retriable_status.cc
@@ -0,0 +1,257 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry for non-retryable status codes.
+// - 1 retry allowed for ABORTED status
+// - first attempt gets INVALID_ARGUMENT, so no retry is done
+static void test_retry_non_retriable_status(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_non_retriable_status", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_INVALID_ARGUMENT;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_INVALID_ARGUMENT);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_non_retriable_status(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_non_retriable_status(config);
+}
+
+void retry_non_retriable_status_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_recv_initial_metadata.cc b/test/core/end2end/tests/retry_recv_initial_metadata.cc
new file mode 100644
index 0000000000..14215e449c
--- /dev/null
+++ b/test/core/end2end/tests/retry_recv_initial_metadata.cc
@@ -0,0 +1,268 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that receiving initial metadata commits the call.
+// - 1 retry allowed for ABORTED status
+// - first attempt receives initial metadata before trailing metadata,
+// so no retry is done even though status was ABORTED
+static void test_retry_recv_initial_metadata(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_recv_initial_metadata", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ // Server sends initial metadata in its own batch, before sending
+ // trailing metadata.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_recv_initial_metadata(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_recv_initial_metadata(config);
+}
+
+void retry_recv_initial_metadata_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_recv_message.cc b/test/core/end2end/tests/retry_recv_message.cc
new file mode 100644
index 0000000000..86171fdc01
--- /dev/null
+++ b/test/core/end2end/tests/retry_recv_message.cc
@@ -0,0 +1,261 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that receiving a message commits the call.
+// - 1 retry allowed for ABORTED status
+// - first attempt receives a message and therefore does not retry even
+// though the final status is ABORTED
+static void test_retry_recv_message(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_recv_message", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = response_payload;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_recv_message(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_recv_message(config);
+}
+
+void retry_recv_message_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_server_pushback_delay.cc b/test/core/end2end/tests/retry_server_pushback_delay.cc
new file mode 100644
index 0000000000..1da986041f
--- /dev/null
+++ b/test/core/end2end/tests/retry_server_pushback_delay.cc
@@ -0,0 +1,318 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we honor server push-back delay.
+// - 2 retries allowed for ABORTED status
+// - first attempt gets ABORTED with a long delay
+// - second attempt succeeds
+static void test_retry_server_pushback_delay(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_metadata pushback_md;
+ memset(&pushback_md, 0, sizeof(pushback_md));
+ pushback_md.key = GRPC_MDSTR_GRPC_RETRY_PUSHBACK_MS;
+ pushback_md.value = grpc_slice_from_static_string("2000");
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 3,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_server_pushback_delay", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 1;
+ op->data.send_status_from_server.trailing_metadata = &pushback_md;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ gpr_timespec before_retry = gpr_now(GPR_CLOCK_MONOTONIC);
+
+ grpc_call_unref(s);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_call_details_init(&call_details);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(201));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+ cq_verify(cqv);
+
+ gpr_timespec after_retry = gpr_now(GPR_CLOCK_MONOTONIC);
+ gpr_timespec retry_delay = gpr_time_sub(after_retry, before_retry);
+ // Configured back-off was 1 second, server push-back said 2 seconds.
+ // To avoid flakiness, we allow some fudge factor here.
+ gpr_log(GPR_INFO, "retry delay was {.tv_sec=%" PRId64 ", .tv_nsec=%d}",
+ retry_delay.tv_sec, retry_delay.tv_nsec);
+ GPR_ASSERT(retry_delay.tv_sec >= 1);
+ if (retry_delay.tv_sec == 1) {
+ GPR_ASSERT(retry_delay.tv_nsec >= 900000000);
+ }
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_OK;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_OK);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 0);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_server_pushback_delay(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_server_pushback_delay(config);
+}
+
+void retry_server_pushback_delay_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_server_pushback_disabled.cc b/test/core/end2end/tests/retry_server_pushback_disabled.cc
new file mode 100644
index 0000000000..13d4f01eea
--- /dev/null
+++ b/test/core/end2end/tests/retry_server_pushback_disabled.cc
@@ -0,0 +1,306 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry when disabled by server push-back.
+// - 2 retries allowed for ABORTED status
+// - first attempt gets ABORTED
+// - second attempt gets ABORTED but server push back disables retrying
+static void test_retry_server_pushback_disabled(
+ grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_metadata pushback_md;
+ memset(&pushback_md, 0, sizeof(pushback_md));
+ pushback_md.key = GRPC_MDSTR_GRPC_RETRY_PUSHBACK_MS;
+ pushback_md.value = grpc_slice_from_static_string("-1");
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 3,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f = begin_test(
+ config, "retry_server_pushback_disabled", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ grpc_call_unref(s);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_call_details_init(&call_details);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(201));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 1;
+ op->data.send_status_from_server.trailing_metadata = &pushback_md;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_server_pushback_disabled(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_server_pushback_disabled(config);
+}
+
+void retry_server_pushback_disabled_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_streaming.cc b/test/core/end2end/tests/retry_streaming.cc
new file mode 100644
index 0000000000..e96e57e8bc
--- /dev/null
+++ b/test/core/end2end/tests/retry_streaming.cc
@@ -0,0 +1,424 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests retrying a streaming RPC. This is the same as
+// the basic retry test, except that the client sends two messages on the
+// call before the initial attempt fails.
+// FIXME: We should also test the case where the retry is committed after
+// replaying 1 of 2 previously-completed send_message ops. However,
+// there's no way to trigger that from an end2end test, because the
+// replayed ops happen under the hood -- they are not surfaced to the
+// C-core API, and therefore we have no way to inject the commit at the
+// right point.
+static void test_retry_streaming(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice request2_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_slice request3_payload_slice = grpc_slice_from_static_string("baz");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("quux");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* request2_payload =
+ grpc_raw_byte_buffer_create(&request2_payload_slice, 1);
+ grpc_byte_buffer* request3_payload =
+ grpc_raw_byte_buffer_create(&request3_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* request2_payload_recv = nullptr;
+ grpc_byte_buffer* request3_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 3,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_streaming", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ // Client starts a batch for receiving initial metadata, a message,
+ // and trailing metadata.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ // Client sends initial metadata and a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+ cq_verify(cqv);
+
+ // Server gets a call with received initial metadata.
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ // Server receives a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ // Client sends a second message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request2_payload;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+ cq_verify(cqv);
+
+ // Server receives the second message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request2_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+ cq_verify(cqv);
+
+ // Server sends both initial and trailing metadata.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+ cq_verify(cqv);
+
+ // Clean up from first attempt.
+ grpc_call_unref(s);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_call_details_init(&call_details);
+ GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+ grpc_byte_buffer_destroy(request_payload_recv);
+ request_payload_recv = nullptr;
+ GPR_ASSERT(
+ byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+ grpc_byte_buffer_destroy(request2_payload_recv);
+ request2_payload_recv = nullptr;
+
+ // Server gets a second call (the retry).
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(201));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ // Server receives a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+ cq_verify(cqv);
+
+ // Server receives a second message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request2_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(203), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(203), true);
+ cq_verify(cqv);
+
+ // Client sends a third message and a close.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request3_payload;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+ cq_verify(cqv);
+
+ // Server receives a third message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request3_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(204), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(204), true);
+ cq_verify(cqv);
+
+ // Server receives a close and sends initial metadata, a message, and
+ // trailing metadata.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = response_payload;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ // Returning a retriable code, but because we are also sending a
+ // message, the client will commit instead of retrying again.
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(205), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(205), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(request2_payload);
+ grpc_byte_buffer_destroy(request3_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+ grpc_byte_buffer_destroy(request_payload_recv);
+ GPR_ASSERT(
+ byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+ grpc_byte_buffer_destroy(request2_payload_recv);
+ GPR_ASSERT(
+ byte_buffer_eq_slice(request3_payload_recv, request3_payload_slice));
+ grpc_byte_buffer_destroy(request3_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_streaming(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_streaming(config);
+}
+
+void retry_streaming_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_streaming_after_commit.cc b/test/core/end2end/tests/retry_streaming_after_commit.cc
new file mode 100644
index 0000000000..43eee86d95
--- /dev/null
+++ b/test/core/end2end/tests/retry_streaming_after_commit.cc
@@ -0,0 +1,354 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we can continue to send/recv messages on a streaming call
+// after retries are committed.
+static void test_retry_streaming_after_commit(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice request2_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("baz");
+ grpc_slice response2_payload_slice = grpc_slice_from_static_string("quux");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* request2_payload =
+ grpc_raw_byte_buffer_create(&request2_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* response2_payload =
+ grpc_raw_byte_buffer_create(&response2_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* request2_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_byte_buffer* response2_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 3,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_streaming_after_commit", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ // Client starts a batch for receiving initial metadata and a message.
+ // This will commit retries.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ // Client sends initial metadata and a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+ cq_verify(cqv);
+
+ // Server gets a call with received initial metadata.
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ // Server receives a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ // Server sends initial metadata and a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = response_payload;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+ // Client receives initial metadata and a message.
+ CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+ cq_verify(cqv);
+
+ // Client sends a second message and a close.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request2_payload;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+ cq_verify(cqv);
+
+ // Server receives a second message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request2_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+ cq_verify(cqv);
+
+ // Server receives a close, sends a second message, and sends status.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = response2_payload;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ // Returning a retriable code, but because retries are already
+ // committed, the client will not retry.
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(105), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(105), true);
+ cq_verify(cqv);
+
+ // Client receives a second message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response2_payload_recv;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(5), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(5), true);
+ cq_verify(cqv);
+
+ // Client receives status.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(request2_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(response2_payload);
+ GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+ grpc_byte_buffer_destroy(request_payload_recv);
+ GPR_ASSERT(
+ byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+ grpc_byte_buffer_destroy(request2_payload_recv);
+ GPR_ASSERT(
+ byte_buffer_eq_slice(response_payload_recv, response_payload_slice));
+ grpc_byte_buffer_destroy(response_payload_recv);
+ GPR_ASSERT(
+ byte_buffer_eq_slice(response2_payload_recv, response2_payload_slice));
+ grpc_byte_buffer_destroy(response2_payload_recv);
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_streaming_after_commit(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_streaming_after_commit(config);
+}
+
+void retry_streaming_after_commit_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_streaming_succeeds_before_replay_finished.cc b/test/core/end2end/tests/retry_streaming_succeeds_before_replay_finished.cc
new file mode 100644
index 0000000000..5c92f64805
--- /dev/null
+++ b/test/core/end2end/tests/retry_streaming_succeeds_before_replay_finished.cc
@@ -0,0 +1,400 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we correctly clean up if the second attempt finishes
+// before we have finished replaying all of the send ops.
+static void test_retry_streaming_succeeds_before_replay_finished(
+ grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice request2_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_slice request3_payload_slice = grpc_slice_from_static_string("baz");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("quux");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* request2_payload =
+ grpc_raw_byte_buffer_create(&request2_payload_slice, 1);
+ grpc_byte_buffer* request3_payload =
+ grpc_raw_byte_buffer_create(&request3_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* request2_payload_recv = nullptr;
+ grpc_byte_buffer* request3_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 3,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_streaming", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ // Client starts a batch for receiving initial metadata, a message,
+ // and trailing metadata.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ // Client sends initial metadata and a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(2), true);
+ cq_verify(cqv);
+
+ // Server gets a call with received initial metadata.
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ // Server receives a message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ // Client sends a second message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request2_payload;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+ cq_verify(cqv);
+
+ // Server receives the second message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request2_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+ cq_verify(cqv);
+
+ // Client sends a third message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request3_payload;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+ cq_verify(cqv);
+
+ // Server receives the third message.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request3_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+ cq_verify(cqv);
+
+ // Server sends both initial and trailing metadata.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(105), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(105), true);
+ cq_verify(cqv);
+
+ // Clean up from first attempt.
+ grpc_call_unref(s);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_call_details_init(&call_details);
+ GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+ grpc_byte_buffer_destroy(request_payload_recv);
+ request_payload_recv = nullptr;
+ GPR_ASSERT(
+ byte_buffer_eq_slice(request2_payload_recv, request2_payload_slice));
+ grpc_byte_buffer_destroy(request2_payload_recv);
+ request2_payload_recv = nullptr;
+ GPR_ASSERT(
+ byte_buffer_eq_slice(request3_payload_recv, request3_payload_slice));
+ grpc_byte_buffer_destroy(request3_payload_recv);
+ request3_payload_recv = nullptr;
+
+ // Server gets a second call (the retry).
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(201));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ // Server receives the first message (and does not receive any others).
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &request_payload_recv;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+ cq_verify(cqv);
+
+ // Server sends initial metadata, a message, and trailing metadata.
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = response_payload;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ // Returning a retriable code, but because we are also sending a
+ // message, the client will commit instead of retrying again.
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(205), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(205), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(request2_payload);
+ grpc_byte_buffer_destroy(request3_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_streaming_succeeds_before_replay_finished(
+ grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_streaming_succeeds_before_replay_finished(config);
+}
+
+void retry_streaming_succeeds_before_replay_finished_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_throttled.cc b/test/core/end2end/tests/retry_throttled.cc
new file mode 100644
index 0000000000..8cd08487ea
--- /dev/null
+++ b/test/core/end2end/tests/retry_throttled.cc
@@ -0,0 +1,264 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we don't retry when throttled.
+// - 1 retry allowed for ABORTED status
+// - first attempt gets ABORTED but is over limit, so no retry is done
+static void test_retry_throttled(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ],\n"
+ // A single failure will cause us to be throttled.
+ // (This is not a very realistic config, but it works for the
+ // purposes of this test.)
+ " \"retryThrottling\": {\n"
+ " \"maxTokens\": 2,\n"
+ " \"tokenRatio\": 1.0,\n"
+ " }\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_throttled", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_throttled(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_throttled(config);
+}
+
+void retry_throttled_pre_init(void) {}
diff --git a/test/core/end2end/tests/retry_too_many_attempts.cc b/test/core/end2end/tests/retry_too_many_attempts.cc
new file mode 100644
index 0000000000..5225c9b229
--- /dev/null
+++ b/test/core/end2end/tests/retry_too_many_attempts.cc
@@ -0,0 +1,299 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+#include "test/core/end2end/cq_verifier.h"
+#include "test/core/end2end/tests/cancel_test_helpers.h"
+
+static void* tag(intptr_t t) { return (void*)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+ const char* test_name,
+ grpc_channel_args* client_args,
+ grpc_channel_args* server_args) {
+ grpc_end2end_test_fixture f;
+ gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+ f = config.create_fixture(client_args, server_args);
+ config.init_server(&f, server_args);
+ config.init_client(&f, client_args);
+ return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+ return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now(void) {
+ return n_seconds_from_now(5);
+}
+
+static void drain_cq(grpc_completion_queue* cq) {
+ grpc_event ev;
+ do {
+ ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+ if (!f->server) return;
+ grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+ GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+ grpc_timeout_seconds_to_deadline(5),
+ nullptr)
+ .type == GRPC_OP_COMPLETE);
+ grpc_server_destroy(f->server);
+ f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+ if (!f->client) return;
+ grpc_channel_destroy(f->client);
+ f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+ shutdown_server(f);
+ shutdown_client(f);
+
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
+ grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+// Tests that we stop retrying after the configured number of attempts.
+// - 1 retry allowed for ABORTED status
+// - first attempt gets ABORTED
+// - second attempt gets ABORTED but does not retry
+static void test_retry_too_many_attempts(grpc_end2end_test_config config) {
+ grpc_call* c;
+ grpc_call* s;
+ grpc_op ops[6];
+ grpc_op* op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_call_details call_details;
+ grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
+ grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
+ grpc_byte_buffer* request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_byte_buffer* response_payload =
+ grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+ grpc_byte_buffer* request_payload_recv = nullptr;
+ grpc_byte_buffer* response_payload_recv = nullptr;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+ int was_cancelled = 2;
+ char* peer;
+
+ grpc_arg arg;
+ arg.type = GRPC_ARG_STRING;
+ arg.key = const_cast<char*>(GRPC_ARG_SERVICE_CONFIG);
+ arg.value.string = const_cast<char*>(
+ "{\n"
+ " \"methodConfig\": [ {\n"
+ " \"name\": [\n"
+ " { \"service\": \"service\", \"method\": \"method\" }\n"
+ " ],\n"
+ " \"retryPolicy\": {\n"
+ " \"maxAttempts\": 2,\n"
+ " \"initialBackoff\": \"1s\",\n"
+ " \"maxBackoff\": \"120s\",\n"
+ " \"backoffMultiplier\": 1.6,\n"
+ " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
+ " }\n"
+ " } ]\n"
+ "}");
+ grpc_channel_args client_args = {1, &arg};
+ grpc_end2end_test_fixture f =
+ begin_test(config, "retry_too_many_attempts", &client_args, nullptr);
+
+ cq_verifier* cqv = cq_verifier_create(f.cq);
+
+ gpr_timespec deadline = five_seconds_from_now();
+ c = grpc_channel_create_call(
+ f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+ grpc_slice_from_static_string("/service/method"),
+ get_host_override_slice("foo.test.google.fr:1234", config), deadline,
+ nullptr);
+ GPR_ASSERT(c);
+
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+ gpr_free(peer);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_slice status_details = grpc_slice_from_static_string("xyz");
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(101));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+ cq_verify(cqv);
+
+ grpc_call_unref(s);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_call_details_init(&call_details);
+
+ error =
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.cq, f.cq, tag(201));
+ GPR_ASSERT(GRPC_CALL_OK == error);
+ CQ_EXPECT_COMPLETION(cqv, tag(201), true);
+ cq_verify(cqv);
+
+ peer = grpc_call_get_peer(s);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+ gpr_free(peer);
+ peer = grpc_call_get_peer(c);
+ GPR_ASSERT(peer != nullptr);
+ gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+ gpr_free(peer);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
+ op->data.send_status_from_server.status_details = &status_details;
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+ error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(202), nullptr);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ CQ_EXPECT_COMPLETION(cqv, tag(202), true);
+ CQ_EXPECT_COMPLETION(cqv, tag(1), true);
+ cq_verify(cqv);
+
+ GPR_ASSERT(status == GRPC_STATUS_ABORTED);
+ GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+ GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/service/method"));
+ validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+ config);
+ GPR_ASSERT(0 == call_details.flags);
+ GPR_ASSERT(was_cancelled == 1);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload);
+ grpc_byte_buffer_destroy(request_payload_recv);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
+ grpc_call_unref(c);
+ grpc_call_unref(s);
+
+ cq_verifier_destroy(cqv);
+
+ end_test(&f);
+ config.tear_down_data(&f);
+}
+
+void retry_too_many_attempts(grpc_end2end_test_config config) {
+ GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
+ test_retry_too_many_attempts(config);
+}
+
+void retry_too_many_attempts_pre_init(void) {}
diff --git a/test/core/transport/BUILD b/test/core/transport/BUILD
index b31d4ff899..2c2d05b9ae 100644
--- a/test/core/transport/BUILD
+++ b/test/core/transport/BUILD
@@ -119,3 +119,15 @@ grpc_cc_test(
"//test/core/util:grpc_test_util",
],
)
+
+grpc_cc_test(
+ name = "status_metadata_test",
+ srcs = ["status_metadata_test.cc"],
+ language = "C++",
+ deps = [
+ "//:grpc",
+ ],
+ external_deps = [
+ "gtest",
+ ],
+)
diff --git a/test/core/transport/status_metadata_test.cc b/test/core/transport/status_metadata_test.cc
new file mode 100644
index 0000000000..a96f11c1ea
--- /dev/null
+++ b/test/core/transport/status_metadata_test.cc
@@ -0,0 +1,61 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "src/core/lib/transport/status_metadata.h"
+
+#include <gtest/gtest.h>
+
+#include "src/core/lib/transport/static_metadata.h"
+
+namespace {
+
+TEST(GetStatusCodeFromMetadata, OK) {
+ EXPECT_EQ(GRPC_STATUS_OK,
+ grpc_get_status_code_from_metadata(GRPC_MDELEM_GRPC_STATUS_0));
+}
+
+TEST(GetStatusCodeFromMetadata, CANCELLED) {
+ EXPECT_EQ(GRPC_STATUS_CANCELLED,
+ grpc_get_status_code_from_metadata(GRPC_MDELEM_GRPC_STATUS_1));
+}
+
+TEST(GetStatusCodeFromMetadata, UNKNOWN) {
+ EXPECT_EQ(GRPC_STATUS_UNKNOWN,
+ grpc_get_status_code_from_metadata(GRPC_MDELEM_GRPC_STATUS_2));
+}
+
+TEST(GetStatusCodeFromMetadata, Other) {
+ grpc_mdelem status_md = grpc_mdelem_from_slices(
+ GRPC_MDSTR_GRPC_STATUS, grpc_slice_from_static_string("10"));
+ EXPECT_EQ(GRPC_STATUS_ABORTED, grpc_get_status_code_from_metadata(status_md));
+ GRPC_MDELEM_UNREF(status_md);
+}
+
+TEST(GetStatusCodeFromMetadata, Unparseable) {
+ grpc_mdelem status_md = grpc_mdelem_from_slices(
+ GRPC_MDSTR_GRPC_STATUS, grpc_slice_from_static_string("NaN"));
+ EXPECT_EQ(GRPC_STATUS_UNKNOWN, grpc_get_status_code_from_metadata(status_md));
+ GRPC_MDELEM_UNREF(status_md);
+}
+
+} // namespace
+
+int main(int argc, char** argv) {
+ ::testing::InitGoogleTest(&argc, argv);
+ return RUN_ALL_TESTS();
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-22 16:55:54 +0000