Fix http client cancel after request finished

У меня падает в этом месте - <https://nda.ya.ru/t/rfmGIXUs7GUqUV> при таком кейсе

1. Запрос завершился успешно
2. Выполняю операцию Cancel() для токена

Если я правильно понял, то падает при обращении к висячему указателю. Видимо проблема в захвате по ссылке - после выхода из функции cancellationEndEvent подыхает
commit_hash:10dd8d3d311e85e6018e8f0ff40806ab82eabbd4

2 files changed, 53 insertions, 7 deletions

diff --git a/library/cpp/http/simple/http_client.h b/library/cpp/http/simple/http_client.h
index 87d3dc095af..d208e0ae055 100644
--- a/library/cpp/http/simple/http_client.h
+++ b/library/cpp/http/simple/http_client.h
@@ -282,10 +282,14 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestReliable(const T&
         const bool haveNewConnection = CreateNewConnectionIfNeeded();
         const bool couldRetry = !haveNewConnection && i == 0; // Actually old connection could be already closed by server,
                                                               // so we should try one more time in this case.
-        TManualEvent cancellationEndEvent;
-        cancellation.Future().Subscribe([&](auto&) {
-            Connection->Shutdown();
-            cancellationEndEvent.Signal();
+        TAtomicSharedPtr<TManualEvent> cancellationEndEvent = MakeAtomicShared<TManualEvent>();
+        auto cancelSub = cancellation.Future().Subscribe([this, cancellationEndEvent](auto&) {
+            if (cancellationEndEvent.RefCount() > 1) {
+                if (Connection && Connection->IsOk()) {
+                    Connection->Shutdown();
+                }
+                cancellationEndEvent->Signal();
+            }
         });
 
         try {
@@ -298,7 +302,7 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestReliable(const T&
             return code;
         } catch (const TSystemError& e) {
             if (cancellation.IsCancellationRequested()) {
-                cancellationEndEvent.WaitI();
+                cancellationEndEvent->WaitI();
                 cancellation.ThrowIfCancellationRequested();
             }
             Connection.Reset();
@@ -307,7 +311,7 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestReliable(const T&
             }
         } catch (const THttpReadException&) { // Actually old connection is already closed by server
             if (cancellation.IsCancellationRequested()) {
-                cancellationEndEvent.WaitI();
+                cancellationEndEvent->WaitI();
                 cancellation.ThrowIfCancellationRequested();
             }
             Connection.Reset();
@@ -316,7 +320,7 @@ TKeepAliveHttpClient::THttpCode TKeepAliveHttpClient::DoRequestReliable(const T&
             }
         } catch (const std::exception&) {
             if (cancellation.IsCancellationRequested()) {
-                cancellationEndEvent.WaitI();
+                cancellationEndEvent->WaitI();
                 cancellation.ThrowIfCancellationRequested();
             }
             Connection.Reset();
diff --git a/library/cpp/http/simple/ut/http_ut.cpp b/library/cpp/http/simple/ut/http_ut.cpp
index 20bef65a09d..e175f1e795c 100644
--- a/library/cpp/http/simple/ut/http_ut.cpp
+++ b/library/cpp/http/simple/ut/http_ut.cpp
@@ -6,6 +6,9 @@
 #include <library/cpp/testing/unittest/registar.h>
 #include <library/cpp/testing/unittest/tests_data.h>
 
+#include <library/cpp/threading/future/async.h>
+#include <util/thread/pool.h>
+
 #include <util/system/event.h>
 #include <util/system/thread.h>
 
@@ -264,6 +267,45 @@ Y_UNIT_TEST_SUITE(SimpleHttp) {
         }
     }
 
+    Y_UNIT_TEST(simpleCancel) {
+        TPortManager pm;
+        ui16 port = pm.GetPort(80);
+        NMock::TMockServer server(createOptions(port, false), []() { return new TPong(TDuration::Seconds(1)); });
+
+        TSimpleHttpClient cl("localhost", port);
+        UNIT_ASSERT_VALUES_EQUAL(0, server.GetClientCount());
+
+        auto tp = CreateThreadPool(3);
+
+        {
+            TStringStream s;
+            NThreading::TCancellationTokenSource cancel;
+            UNIT_ASSERT_NO_EXCEPTION(cl.DoGet("/ping", &s, TKeepAliveHttpClient::THeaders(), nullptr, cancel.Token()));
+            cancel.Cancel();
+            UNIT_ASSERT_VALUES_EQUAL("pong", s.Str());
+            Sleep(TDuration::MilliSeconds(500));
+            UNIT_ASSERT_VALUES_EQUAL(0, server.GetClientCount());
+        }
+
+        {
+            TStringStream s;
+            NThreading::TCancellationTokenSource cancel;
+            auto reqFuture = NThreading::Async([&] {
+                // Если DoGet() при отмене кидает исключение — оно “переедет” в future.
+                return cl.DoGet("/ping",
+                                &s,
+                                TKeepAliveHttpClient::THeaders(),
+                                nullptr,
+                                cancel.Token());
+            }, *tp);
+            Sleep(TDuration::MilliSeconds(50));
+            cancel.Cancel();
+            UNIT_ASSERT_EXCEPTION(reqFuture.GetValueSync(), NThreading::TOperationCancelledException);
+            Sleep(TDuration::MilliSeconds(1000));
+            UNIT_ASSERT_VALUES_EQUAL(0, server.GetClientCount());
+        }
+    }
+
     Y_UNIT_TEST(simpleMessages) {
         TPortManager pm;
         ui16 port = pm.GetPort(80);