Issue-3: reject requests to unavailable agents in RDMA partition

cloud/blockstore/libs/endpoint_proxy/server/bootstrap.cpp

@@ -30,6 +30,7 @@ void TBootstrap::Init()
         Options.StoredEndpointsPath,
         Options.NbdRequestTimeout,
         Options.NbdReconnectDelay,
+        Options.DebugRestartEventsCount,
     },
     CreateWallClockTimer(),
     Scheduler,
@@ -44,8 +45,12 @@ void TBootstrap::Start()
 
 void TBootstrap::Stop()
 {
-    Server->Stop();
-    Scheduler->Stop();
+    if (Server) {
+        Server->Stop();
+    }
+    if (Scheduler) {
+        Scheduler->Stop();
+    }
 }
 
 }   // namespace NCloud::NBlockStore::NServer

cloud/blockstore/libs/endpoint_proxy/server/options.cpp

@@ -33,13 +33,13 @@ TOptions::TOptions()
         .StoreResult(&StoredEndpointsPath);
 
     Opts.AddLongOption("nbd-request-timeout")
-        .OptionalArgument("NUM")
+        .RequiredArgument("NUM")
         .Handler1T<TString>([this] (const auto& s) {
             NbdRequestTimeout = TDuration::Parse(s);
         });
 
     Opts.AddLongOption("nbd-reconnect-delay")
-        .OptionalArgument("NUM")
+        .RequiredArgument("NUM")
         .Handler1T<TString>([this] (const auto& s) {
             NbdReconnectDelay = TDuration::Parse(s);
         });
@@ -48,6 +48,13 @@ TOptions::TOptions()
     Opts.AddLongOption("without-libnl")
         .NoArgument()
         .SetFlag(&WithoutLibnl);
+
+    Opts.AddLongOption(
+            "debug-restart-events-count",
+            "issue multiple restart events for each io error. this option is "
+            "used to debug nbd server reconnect")
+        .RequiredArgument("NUM")
+        .StoreResult(&DebugRestartEventsCount);
 }
 
 void TOptions::Parse(int argc, char** argv)

cloud/blockstore/libs/endpoint_proxy/server/options.h

@@ -21,6 +21,7 @@ struct TOptions final: TOptionsBase
     TDuration NbdRequestTimeout = TDuration::Minutes(10);
     TDuration NbdReconnectDelay = TDuration::MilliSeconds(100);
     bool WithoutLibnl = false;
+    std::optional<ui32> DebugRestartEventsCount;
 
     TOptions();
 

cloud/blockstore/libs/endpoint_proxy/server/server.cpp

@@ -79,12 +79,11 @@ struct TEndpoint: std::enable_shared_from_this<TEndpoint>
     NBD::IDevicePtr NbdDevice;
     std::unique_ptr<TNetworkAddress> ListenAddress;
     IProxyRequestStatsPtr RequestStats;
-
     TString UnixSocketPath;
     TString InternalUnixSocketPath;
     TString NbdDevicePath;
-
     NBD::TStorageOptions NbdOptions;
+    std::atomic<ui64> Generation = 0;
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -191,16 +190,19 @@ struct TRestartAlarmContext: TRequestContextBase
 {
     std::weak_ptr<TEndpoint> Endpoint;
     grpc::ServerCompletionQueue& CQ;
+    const ui64 Generation;
     TBackoffDelayProvider Backoff;
     grpc::Alarm Alarm;
 
     TRestartAlarmContext(
             std::weak_ptr<TEndpoint> endpoint,
             grpc::ServerCompletionQueue& cq,
+            ui64 generation,
             TDuration minReconnectDelay,
             TDuration maxReconnectDelay)
         : Endpoint(std::move(endpoint))
         , CQ(cq)
+        , Generation(generation)
         , Backoff{minReconnectDelay, maxReconnectDelay}
     {
         SetAlarm();
@@ -275,27 +277,36 @@ struct TServer: IEndpointProxyServer
     {
         std::weak_ptr<TEndpoint> Endpoint;
         grpc::ServerCompletionQueue& CQ;
+        const ui64 Generation;
         TDuration ReconnectDelay;
+        std::optional<ui32> DebugRestartEventsCount;
 
         TErrorHandler(
                 std::weak_ptr<TEndpoint> endpoint,
                 grpc::ServerCompletionQueue& cq,
-                TDuration reconnectDelay)
+                ui64 generation,
+                TDuration reconnectDelay,
+                std::optional<ui32> debugRestartEventsCount)
             : Endpoint(std::move(endpoint))
             , CQ(cq)
+            , Generation(generation)
             , ReconnectDelay(reconnectDelay)
+            , DebugRestartEventsCount(debugRestartEventsCount)
         {}
 
         void ProcessException(std::exception_ptr e) override
         {
             Y_UNUSED(e);
 
-            // context will be held by grpc until CQ->Next returns it
-            new TRestartAlarmContext(
-                Endpoint,
-                CQ,
-                ReconnectDelay,
-                MAX_RECONNECT_DELAY);
+            for (ui32 i = 0; i < DebugRestartEventsCount.value_or(1); i++) {
+                // context will be held by grpc until CQ->Next returns it
+                new TRestartAlarmContext(
+                    Endpoint,
+                    CQ,
+                    Generation,
+                    ReconnectDelay,
+                    MAX_RECONNECT_DELAY);
+            }
         }
     };
 
@@ -755,9 +766,16 @@ struct TServer: IEndpointProxyServer
         response.SetNbdDevice(ep.NbdDevicePath);
 
         if (ep.NbdDevice) {
-            ep.NbdDevice->Stop(true);
-            STORAGE_INFO(request.ShortDebugString().Quote()
-                << " - Stopped NBD device");
+            auto err = ep.NbdDevice->Stop(true).GetValueSync();
+            if (HasError(err)) {
+                STORAGE_ERROR(
+                    request.ShortDebugString().Quote()
+                    << " - Failed to stop NBD device: " << err.GetCode());
+            } else {
+                STORAGE_INFO(
+                    request.ShortDebugString().Quote()
+                    << " - Stopped NBD device");
+            }
         }
 
         if (ep.ListenAddress) {
@@ -945,7 +963,9 @@ struct TServer: IEndpointProxyServer
             std::make_shared<TErrorHandler>(
                 ep.shared_from_this(),
                 *CQ,
-                Config.NbdReconnectDelay),
+                ep.Generation,
+                Config.NbdReconnectDelay,
+                Config.DebugRestartEventsCount),
             ep.NbdOptions);
 
         // TODO fix StartEndpoint signature - it's actually synchronous
@@ -1006,6 +1026,11 @@ struct TServer: IEndpointProxyServer
     bool ProcessAlarm(TRestartAlarmContext* context)
     {
         if (auto ep = context->Endpoint.lock()) {
+            if (context->Generation != ep->Generation) {
+                return false;
+            }
+            ep->Generation++;
+
             const auto tag = TStringBuilder()
                 << "UnixSocketPath: " << ep->UnixSocketPath.Quote() << " - ";
 
@@ -1025,8 +1050,13 @@ struct TServer: IEndpointProxyServer
     bool DoProcessAlarm(TEndpoint& ep, const TString& tag)
     {
         if (ep.NbdDevice) {
-            ep.NbdDevice->Stop(NBD_DELETE_DEVICE).GetValueSync();
-            STORAGE_INFO(tag << "Stopped NBD device");
+            auto err = ep.NbdDevice->Stop(NBD_DELETE_DEVICE).GetValueSync();
+            if (HasError(err)) {
+                STORAGE_ERROR(
+                    tag << "Failed to stop NBD device: " << err.GetCode());
+            } else {
+                STORAGE_INFO(tag << "Stopped NBD device");
+            }
         }
 
         if (ep.ListenAddress) {

cloud/blockstore/libs/endpoint_proxy/server/server.h

@@ -29,6 +29,7 @@ struct TEndpointProxyServerConfig
     TString StoredEndpointsPath;
     TDuration NbdRequestTimeout;
     TDuration NbdReconnectDelay;
+    std::optional<ui32> DebugRestartEventsCount;
 
     TEndpointProxyServerConfig(
             ui16 port,
@@ -40,7 +41,8 @@ struct TEndpointProxyServerConfig
             bool netlink,
             TString storedEndpointsPath,
             TDuration nbdRequestTimeout,
-            TDuration nbdReconnectDelay)
+            TDuration nbdReconnectDelay,
+            std::optional<ui32> debugRestartEventsCount)
         : Port(port)
         , SecurePort(securePort)
         , RootCertsFile(std::move(rootCertsFile))
@@ -51,7 +53,9 @@ struct TEndpointProxyServerConfig
         , StoredEndpointsPath(std::move(storedEndpointsPath))
         , NbdRequestTimeout(nbdRequestTimeout)
         , NbdReconnectDelay(nbdReconnectDelay)
+        , DebugRestartEventsCount(debugRestartEventsCount)
     {
+        Y_ENSURE(debugRestartEventsCount.value_or(1) > 0);
     }
 };
 

cloud/blockstore/libs/endpoint_proxy/server/server_ut.cpp

@@ -59,7 +59,8 @@ Y_UNIT_TEST_SUITE(TServerTest)
             false,
             "",
             TDuration::Seconds(1),
-            TDuration::MilliSeconds(100));
+            TDuration::MilliSeconds(100),
+            1);
 
         auto server = CreateServer(
             serverConfig,

cloud/blockstore/libs/nbd/netlink_device.cpp

@@ -338,7 +338,7 @@ void TNetlinkDevice::Disconnect()
         socket.Receive(response);
         StopResult.SetValue(MakeError(S_OK));
     } catch (const TServiceError& e) {
-        StartResult.SetValue(MakeError(
+        StopResult.SetValue(MakeError(
             e.GetCode(),
             TStringBuilder()
                 << "unable to disconnect " << DeviceName << ": " << e.what()));

cloud/blockstore/libs/rdma/iface/client.h

@@ -103,10 +103,13 @@ struct IClientEndpoint
         size_t requestBytes,
         size_t responseBytes) = 0;
 
-    virtual void SendRequest(
+    // Returns id of sended request. It can be used to cancel this request.
+    virtual ui64 SendRequest(
         TClientRequestPtr req,
         TCallContextPtr callContext) = 0;
 
+    virtual void CancelRequest(ui64 reqId) = 0;
+
     virtual NThreading::TFuture<void> Stop() = 0;
 };
 

cloud/blockstore/libs/rdma/iface/protocol.h

@@ -36,6 +36,7 @@ enum {
     RDMA_PROTO_CONFIG_MISMATCH  = 2,
     RDMA_PROTO_THROTTLED        = 3,
     RDMA_PROTO_FAIL             = 4,
+    RDMA_PROTO_CANCELLED        = 5,
 };
 
 enum {