SERVER-58024 Audit NetworkInterfaceTL for calls to exception-throwing fns in getAsync continuations

3 files changed, 92 insertions, 10 deletions

diff --git a/src/mongo/executor/SConscript b/src/mongo/executor/SConscript
index a382c094de8..627675845a2 100644
--- a/src/mongo/executor/SConscript
+++ b/src/mongo/executor/SConscript
@@ -156,6 +156,7 @@ env.Library(
     source=[
         'connection_pool_tl.cpp',
         'network_interface_tl.cpp',
+        'network_interface_tl.idl',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/client/async_client',
diff --git a/src/mongo/executor/network_interface_tl.cpp b/src/mongo/executor/network_interface_tl.cpp
index 5fa9a00e6cd..1142ef6eec5 100644
--- a/src/mongo/executor/network_interface_tl.cpp
+++ b/src/mongo/executor/network_interface_tl.cpp
@@ -29,16 +29,17 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kASIO
 
-#include "mongo/platform/basic.h"
-
 #include "mongo/executor/network_interface_tl.h"
 
+#include <fmt/format.h>
+
 #include "mongo/config.h"
 #include "mongo/db/auth/security_token.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/wire_version.h"
 #include "mongo/executor/connection_pool_tl.h"
 #include "mongo/executor/hedging_metrics.h"
+#include "mongo/executor/network_interface_tl_gen.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/transport/transport_layer_manager.h"
@@ -46,10 +47,11 @@
 #include "mongo/util/net/socket_utils.h"
 #include "mongo/util/testing_proctor.h"
 
-
 namespace mongo {
 namespace executor {
 
+using namespace fmt::literals;
+
 namespace {
 static inline const std::string kMaxTimeMSOpOnlyField = "maxTimeMSOpOnly";
 
@@ -73,6 +75,33 @@ Status appendMetadata(RemoteCommandRequestOnAny* request,
 
     return Status::OK();
 }
+
+/**
+ * Invokes `f()`, and returns true if it succeeds.
+ * Otherwise, we log the exception with a `hint` string and handle the error.
+ * The exception handling has two possibilities, controlled by the server parameter
+ * `suppressNetworkInterfaceTransportLayerExceptions`.
+ * The old behavior is to simply rethrow the exception, which will crash the process.
+ * The new behavior is to invoke `eh(err)` and return false. This gives the caller a way
+ * to provide a route to propagate the exception as a Status (perhaps filling a promise
+ * with it) and carry on.
+ */
+template <typename F, typename EH>
+bool catchingInvoke(F&& f, EH&& eh, StringData hint) {
+    try {
+        std::forward<F>(f)();
+        return true;
+    } catch (...) {
+        Status err = exceptionToStatus();
+        LOGV2(5802401, "Callback failed", "msg"_attr = hint, "error"_attr = err);
+        if (gSuppressNetworkInterfaceTransportLayerExceptions.isEnabledAndIgnoreFCV())
+            std::forward<EH>(eh)(err);  // new server parameter protected behavior
+        else
+            throw;  // old behavior
+        return false;
+    }
+}
+
 }  // namespace
 
 /**
@@ -530,7 +559,9 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa
         .thenRunOn(makeGuaranteedExecutor(baton, _reactor))
         .getAsync([cmdState = cmdState,
                    onFinish = std::move(onFinish)](StatusWith<RemoteCommandOnAnyResponse> swr) {
-            invariant(swr.isOK());
+            invariant(swr.isOK(),
+                      "Remote command response failed with an error: {}"_format(
+                          swr.getStatus().toString()));
             auto rs = std::move(swr.getValue());
             // The TransportLayer has, for historical reasons returned
             // SocketException for network errors, but sharding assumes
@@ -546,7 +577,9 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa
                         "isOK"_attr = rs.isOK(),
                         "response"_attr =
                             redact(rs.isOK() ? rs.data.toString() : rs.status.toString()));
-            onFinish(std::move(rs));
+            catchingInvoke([&] { onFinish(std::move(rs)); },
+                           [&](Status& err) { cmdState->fulfillFinalPromise(err); },
+                           "The finish callback failed. Aborting exhaust command");
         });
 
     if (MONGO_unlikely(networkInterfaceDiscardCommandsBeforeAcquireConn.shouldFail())) {
@@ -596,7 +629,10 @@ Future<RemoteCommandResponse> NetworkInterfaceTL::CommandState::sendRequest(
                    ->runCommandRequest(*requestState->request, baton);
            })
         .then([this, requestState](RemoteCommandResponse response) {
-            doMetadataHook(RemoteCommandOnAnyResponse(requestState->host, response));
+            catchingInvoke(
+                [&] { doMetadataHook(RemoteCommandOnAnyResponse(requestState->host, response)); },
+                [&](Status& err) { promise.setError(err); },
+                "Metadata hook readReplyMetadata");
             return response;
         });
 }
@@ -979,7 +1015,10 @@ void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest(
     }
 
     auto onAnyResponse = RemoteCommandOnAnyResponse(requestState->host, response);
-    doMetadataHook(onAnyResponse);
+    if (!catchingInvoke([&] { doMetadataHook(onAnyResponse); },
+                        [&](Status& err) { finalResponsePromise.setError(err); },
+                        "Exhaust command metadata hook readReplyMetadata"))
+        return;
 
     // If the command failed, we will call 'onReply' as a part of the future chain paired with
     // the promise. This is to be sure that all error paths will run 'onReply' only once upon
@@ -992,14 +1031,20 @@ void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest(
         return;
     }
 
-    onReplyFn(onAnyResponse);
+    if (!catchingInvoke([&] { onReplyFn(onAnyResponse); },
+                        [&](Status& err) { finalResponsePromise.setError(err); },
+                        "Exhaust command onReplyFn"))
+        return;
 
-    // Reset the stopwatch to measure the correct duration for the folowing reply
+    // Reset the stopwatch to measure the correct duration for the following reply
     stopwatch.restart();
     if (deadline != kNoExpirationDate) {
         deadline = stopwatch.start() + requestOnAny.timeout;
     }
-    setTimer();
+    if (!catchingInvoke([&] { setTimer(); },
+                        [&](Status& err) { finalResponsePromise.setError(err); },
+                        "Exhaust command setTimer"))
+        return;
 
     requestState->getClient(requestState->conn)
         ->awaitExhaustCommand(baton)
diff --git a/src/mongo/executor/network_interface_tl.idl b/src/mongo/executor/network_interface_tl.idl
new file mode 100644
index 00000000000..b71850511f4
--- /dev/null
+++ b/src/mongo/executor/network_interface_tl.idl
@@ -0,0 +1,36 @@
+# Copyright (C) 2022-present MongoDB, Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the Server Side Public License, version 1,
+# as published by MongoDB, Inc.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# Server Side Public License for more details.
+#
+# You should have received a copy of the Server Side Public License
+# along with this program. If not, see
+# <http://www.mongodb.com/licensing/server-side-public-license>.
+#
+# As a special exception, the copyright holders give permission to link the
+# code of portions of this program with the OpenSSL library under certain
+# conditions as described in each individual source file and distribute
+# linked combinations including the program with the OpenSSL library. You
+# must comply with the Server Side Public License in all respects for
+# all of the code used other than as permitted herein. If you modify file(s)
+# with this exception, you may extend this exception to your version of the
+# file(s), but you are not obligated to do so. If you do not wish to do so,
+# delete this exception statement from your version. If you delete this
+# exception statement from all source files in the program, then also delete
+# it in the license file.
+#
+
+global:
+  cpp_namespace: "mongo::executor"
+
+feature_flags:
+  suppressNetworkInterfaceTransportLayerExceptions:
+    description: Suppress network interface transport layer exceptions
+    cpp_varname: gSuppressNetworkInterfaceTransportLayerExceptions
+    default: false