|
17 | 17 | #include <atomic> |
18 | 18 | #include <chrono> |
19 | 19 | #include <condition_variable> |
| 20 | +#include <cstdint> |
20 | 21 | #include <cstdio> |
21 | 22 | #include <cstring> |
| 23 | +#include <limits> |
22 | 24 | #include <map> |
23 | 25 | #include <memory> |
24 | 26 | #include <mutex> |
@@ -120,6 +122,37 @@ class CustomEventHandler : public http_client::EventHandler |
120 | 122 | // inside the Response event, which is the one moment both arms of the completion callback are |
121 | 123 | // eligible: DispatchEvent notifies the handler before it stores the new state, and the callback |
122 | 124 | // runs after both, so it sees an aborted operation that also has a response. |
| 125 | +class ReentrantSendHandler : public CustomEventHandler |
| 126 | +{ |
| 127 | +public: |
| 128 | + void OnResponse(http_client::Response & /* response */) noexcept override |
| 129 | + { |
| 130 | + got_response_.store(true, std::memory_order_release); |
| 131 | + |
| 132 | + if (session_ != nullptr && !resent_.exchange(true, std::memory_order_acq_rel)) |
| 133 | + { |
| 134 | + auto again = session_->CreateRequest(); |
| 135 | + again->SetUri("get/"); |
| 136 | + session_->SendRequest(self_.lock()); |
| 137 | + } |
| 138 | + } |
| 139 | + |
| 140 | + void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override |
| 141 | + { |
| 142 | + if (state == http_client::SessionState::CreateFailed) |
| 143 | + { |
| 144 | + create_failed_.fetch_add(1, std::memory_order_release); |
| 145 | + } |
| 146 | + } |
| 147 | + |
| 148 | + http_client::Session *session_ = nullptr; |
| 149 | + std::weak_ptr<ReentrantSendHandler> self_; |
| 150 | + std::atomic<int> create_failed_{0}; |
| 151 | + |
| 152 | +private: |
| 153 | + std::atomic<bool> resent_{false}; |
| 154 | +}; |
| 155 | + |
123 | 156 | class TerminalCountingHandler : public CustomEventHandler |
124 | 157 | { |
125 | 158 | public: |
@@ -262,6 +295,7 @@ class BasicCurlHttpTests : public ::testing::Test, public HTTP_SERVER_NS::HttpRe |
262 | 295 | server_.addHandler("/get/", *this); |
263 | 296 | server_.addHandler("/post/", *this); |
264 | 297 | server_.addHandler("/retry/", *this); |
| 298 | + server_.addHandler("/retry-after/", *this); |
265 | 299 | server_.addHandler("/close/", *this); |
266 | 300 | server_.start(); |
267 | 301 | is_running_ = true; |
@@ -302,6 +336,14 @@ class BasicCurlHttpTests : public ::testing::Test, public HTTP_SERVER_NS::HttpRe |
302 | 336 | response.headers["Content-Type"] = "text/plain"; |
303 | 337 | response_status = 429; |
304 | 338 | } |
| 339 | + else if (request.uri == "/retry-after/") |
| 340 | + { |
| 341 | + std::unique_lock<std::mutex> lk1(mtx_requests); |
| 342 | + received_requests_.push_back(request); |
| 343 | + response.headers["Content-Type"] = "text/plain"; |
| 344 | + response.headers["Retry-After"] = "30"; |
| 345 | + response_status = 429; |
| 346 | + } |
305 | 347 | else if (request.uri == "/close/") |
306 | 348 | { |
307 | 349 | // -1 is the documented way for a handler to ask the server to terminate the |
@@ -506,6 +548,12 @@ TEST_F(BasicCurlHttpTests, SeekCallbackRepositionsTheRequestBody) |
506 | 548 | EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::Seek(operation, -1, SEEK_SET)); |
507 | 549 | EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::Seek(operation, 0, SEEK_CUR)); |
508 | 550 | EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::Seek(operation, 0, SEEK_END)); |
| 551 | + |
| 552 | + // An offset past the range of size_t must stay out of range. Narrowing it to size_t first |
| 553 | + // wrapped it back into the body on a build where size_t is 32 bits. |
| 554 | + EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, |
| 555 | + Peer::Seek(operation, static_cast<curl_off_t>(std::numeric_limits<uint32_t>::max()) + 4, |
| 556 | + SEEK_SET)); |
509 | 557 | EXPECT_EQ(3u, Peer::ReadCursor(operation)); |
510 | 558 |
|
511 | 559 | EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::SeekNullUserData(0, SEEK_SET)); |
@@ -638,6 +686,90 @@ TEST_F(BasicCurlHttpTests, ExponentialBackoffRetry) |
638 | 686 | ASSERT_EQ(CURLE_OK, operation.Send()); |
639 | 687 | ASSERT_FALSE(operation.IsRetryable()); |
640 | 688 | } |
| 689 | + |
| 690 | +// A Retry-After beyond max_backoff closes the session. The IO loop used to queue it anyway, where |
| 691 | +// it held back later retries and the background thread until the server's time. |
| 692 | +TEST_F(BasicCurlHttpTests, RetryAfterBeyondMaxBackoffIsNotQueued) |
| 693 | +{ |
| 694 | + received_requests_.clear(); |
| 695 | + curl::HttpClient http_client; |
| 696 | + const http_client::RetryPolicy retry_policy = {2, std::chrono::duration<float>{0.1f}, |
| 697 | + std::chrono::duration<float>{1.0f}, 1.0f}; |
| 698 | + |
| 699 | + auto capped_session = http_client.CreateSession("http://127.0.0.1:19000"); |
| 700 | + auto capped_request = capped_session->CreateRequest(); |
| 701 | + capped_request->SetMethod(http_client::Method::Post); |
| 702 | + capped_request->SetUri("retry-after/"); |
| 703 | + capped_request->SetRetryPolicy(retry_policy); |
| 704 | + auto capped_handler = std::make_shared<RetryEventHandler>(); |
| 705 | + capped_session->SendRequest(capped_handler); |
| 706 | + capped_session->FinishSession(); |
| 707 | + ASSERT_TRUE(capped_handler->got_response_.load(std::memory_order_acquire)); |
| 708 | + |
| 709 | + auto session = http_client.CreateSession("http://127.0.0.1:19000"); |
| 710 | + auto request = session->CreateRequest(); |
| 711 | + request->SetMethod(http_client::Method::Post); |
| 712 | + request->SetUri("retry/"); |
| 713 | + request->SetRetryPolicy(retry_policy); |
| 714 | + auto handler = std::make_shared<RetryEventHandler>(); |
| 715 | + auto started_at = std::chrono::steady_clock::now(); |
| 716 | + session->SendRequest(handler); |
| 717 | + session->FinishSession(); |
| 718 | + const auto retried_in = std::chrono::steady_clock::now() - started_at; |
| 719 | + ASSERT_TRUE(handler->got_response_.load(std::memory_order_acquire)); |
| 720 | + |
| 721 | + started_at = std::chrono::steady_clock::now(); |
| 722 | + http_client.WaitBackgroundThreadExit(); |
| 723 | + const auto joined_in = std::chrono::steady_clock::now() - started_at; |
| 724 | + |
| 725 | + // The server asks for 30 s; the policy backs off for about 0.1 s. |
| 726 | + EXPECT_TRUE(retried_in < std::chrono::seconds{10}) |
| 727 | + << "retry ms: " << std::chrono::duration_cast<std::chrono::milliseconds>(retried_in).count(); |
| 728 | + EXPECT_TRUE(joined_in < std::chrono::seconds{10}) |
| 729 | + << "join ms: " << std::chrono::duration_cast<std::chrono::milliseconds>(joined_in).count(); |
| 730 | + |
| 731 | + std::unique_lock<std::mutex> lock_requests(mtx_requests); |
| 732 | + const auto hits = [this](const char *uri) { |
| 733 | + return std::count_if( |
| 734 | + received_requests_.begin(), received_requests_.end(), |
| 735 | + [uri](const HTTP_SERVER_NS::HttpRequest &received) { return received.uri == uri; }); |
| 736 | + }; |
| 737 | + EXPECT_EQ(1, hits("/retry-after/")); |
| 738 | + EXPECT_EQ(2, hits("/retry/")); |
| 739 | +} |
| 740 | + |
| 741 | +// The shutdown half of #4631: the closed session used to hold the join until the server's time. |
| 742 | +TEST_F(BasicCurlHttpTests, RetryAfterBeyondMaxBackoffDoesNotDelayShutdown) |
| 743 | +{ |
| 744 | + received_requests_.clear(); |
| 745 | + curl::HttpClient http_client; |
| 746 | + const http_client::RetryPolicy retry_policy = {2, std::chrono::duration<float>{0.1f}, |
| 747 | + std::chrono::duration<float>{1.0f}, 1.0f}; |
| 748 | + |
| 749 | + auto session = http_client.CreateSession("http://127.0.0.1:19000"); |
| 750 | + auto request = session->CreateRequest(); |
| 751 | + request->SetMethod(http_client::Method::Post); |
| 752 | + request->SetUri("retry-after/"); |
| 753 | + request->SetRetryPolicy(retry_policy); |
| 754 | + auto handler = std::make_shared<RetryEventHandler>(); |
| 755 | + session->SendRequest(handler); |
| 756 | + session->FinishSession(); |
| 757 | + ASSERT_TRUE(handler->got_response_.load(std::memory_order_acquire)); |
| 758 | + |
| 759 | + const auto started_at = std::chrono::steady_clock::now(); |
| 760 | + http_client.WaitBackgroundThreadExit(); |
| 761 | + const auto joined_in = std::chrono::steady_clock::now() - started_at; |
| 762 | + |
| 763 | + // The server asks for 30 s. |
| 764 | + EXPECT_TRUE(joined_in < std::chrono::seconds{10}) |
| 765 | + << "join ms: " << std::chrono::duration_cast<std::chrono::milliseconds>(joined_in).count(); |
| 766 | + |
| 767 | + std::unique_lock<std::mutex> lock_requests(mtx_requests); |
| 768 | + EXPECT_EQ(1, std::count_if(received_requests_.begin(), received_requests_.end(), |
| 769 | + [](const HTTP_SERVER_NS::HttpRequest &received) { |
| 770 | + return received.uri == "/retry-after/"; |
| 771 | + })); |
| 772 | +} |
641 | 773 | #endif // ENABLE_OTLP_RETRY_PREVIEW |
642 | 774 |
|
643 | 775 | // A cancel that arrives once the server has answered used to deliver Cancelled and the response, |
@@ -742,6 +874,63 @@ TEST_F(BasicCurlHttpTests, ResetMultiHandleWithASessionDoesNotDeadlock) |
742 | 874 | client->FinishAllSessions(); |
743 | 875 | } |
744 | 876 |
|
| 877 | +// The reproduction from #4396. A handler that sends again from OnResponse replaces the operation |
| 878 | +// whose Cleanup is running that handler, and Cleanup reads that operation again on the way out. |
| 879 | +TEST_F(BasicCurlHttpTests, ASecondRequestFromInsideTheResponseIsRefused) |
| 880 | +{ |
| 881 | + received_requests_.clear(); |
| 882 | + auto session_manager = std::make_shared<http_client::curl::HttpCurlClientFactory>()->Create(); |
| 883 | + ASSERT_TRUE(session_manager != nullptr); |
| 884 | + |
| 885 | + auto session = session_manager->CreateSession("http://127.0.0.1:19000"); |
| 886 | + auto request = session->CreateRequest(); |
| 887 | + request->SetUri("get/"); |
| 888 | + |
| 889 | + auto handler = std::make_shared<ReentrantSendHandler>(); |
| 890 | + handler->session_ = session.get(); |
| 891 | + handler->self_ = handler; |
| 892 | + |
| 893 | + session->SendRequest(handler); |
| 894 | + ASSERT_TRUE(waitForRequests(30, 1)); |
| 895 | + session->FinishSession(); |
| 896 | + |
| 897 | + EXPECT_TRUE(handler->got_response_.load(std::memory_order_acquire)); |
| 898 | + EXPECT_EQ(1, handler->create_failed_.load(std::memory_order_acquire)); |
| 899 | + |
| 900 | + session_manager->FinishAllSessions(); |
| 901 | +} |
| 902 | + |
| 903 | +// The same thing without the re-entrancy, and the request the first send is reading from is not |
| 904 | +// replaced under it either. |
| 905 | +TEST_F(BasicCurlHttpTests, ASessionSendsOneRequest) |
| 906 | +{ |
| 907 | + received_requests_.clear(); |
| 908 | + auto session_manager = std::make_shared<http_client::curl::HttpCurlClientFactory>()->Create(); |
| 909 | + ASSERT_TRUE(session_manager != nullptr); |
| 910 | + |
| 911 | + auto session = session_manager->CreateSession("http://127.0.0.1:19000"); |
| 912 | + auto request = session->CreateRequest(); |
| 913 | + request->SetUri("get/"); |
| 914 | + |
| 915 | + auto first = std::make_shared<ReentrantSendHandler>(); |
| 916 | + session->SendRequest(first); |
| 917 | + ASSERT_TRUE(waitForRequests(30, 1)); |
| 918 | + session->FinishSession(); |
| 919 | + |
| 920 | + EXPECT_TRUE(first->got_response_.load(std::memory_order_acquire)); |
| 921 | + EXPECT_EQ(0, first->create_failed_.load(std::memory_order_acquire)); |
| 922 | + |
| 923 | + auto again = session->CreateRequest(); |
| 924 | + EXPECT_EQ(request.get(), again.get()); |
| 925 | + |
| 926 | + auto second = std::make_shared<ReentrantSendHandler>(); |
| 927 | + session->SendRequest(second); |
| 928 | + EXPECT_EQ(1, second->create_failed_.load(std::memory_order_acquire)); |
| 929 | + EXPECT_FALSE(second->got_response_.load(std::memory_order_acquire)); |
| 930 | + |
| 931 | + session_manager->FinishAllSessions(); |
| 932 | +} |
| 933 | + |
745 | 934 | // The caller-thread side of the same cancel. The server handler takes mtx_requests before it |
746 | 935 | // answers, so holding it keeps a response from racing the cancel and the abort lands while the |
747 | 936 | // IO thread is still driving the easy handle. That pairing is what #4369 caught. |
|
0 commit comments