net: ss::nested_exception might be a disconnect_exception

src/v/net/connection.cc

@@ -12,6 +12,8 @@
 #include "net/exceptions.h"
 #include "rpc/service.h"
 
+#include <seastar/core/future.hh>
+
 #include <gnutls/gnutls.h>
 
 namespace net {
@@ -87,6 +89,11 @@ std::optional<ss::sstring> is_disconnect_exception(std::exception_ptr e) {
             return fmt::format("invalid request: {}", e.what());
         }
         return "invalid request";
+    } catch (const ss::nested_exception& e) {
+        if (auto err = is_disconnect_exception(e.inner)) {
+            return err;
+        }
+        return is_disconnect_exception(e.outer);
     } catch (...) {
         // Global catch-all prevents stranded/non-handled exceptional futures.
         // In all other non-explicity handled cases, the exception will not be

src/v/net/tests/CMakeLists.txt

@@ -29,3 +29,13 @@ rp_test(
         ARGS "-- -c 1"
         LABELS net
 )
+
+rp_test(
+        UNIT_TEST
+        BINARY_NAME net_connection
+        SOURCES
+        connection.cc
+        DEFINITIONS BOOST_TEST_DYN_LINK
+        LIBRARIES v::seastar_testing_main Boost::unit_test_framework v::net
+        LABELS net
+)

src/v/net/tests/connection.cc

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2023 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#include "net/connection.h"
+
+#include <seastar/core/future.hh>
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#include <exception>
+#include <system_error>
+
+BOOST_AUTO_TEST_CASE(test_is_disconnect_error) {
+    auto is_a_de = std::system_error(
+      std::make_error_code(std::errc::broken_pipe));
+    auto not_a_de = std::system_error(
+      std::make_error_code(std::errc::is_a_directory));
+
+    BOOST_CHECK(net::is_disconnect_exception(std::make_exception_ptr(is_a_de))
+                  .has_value());
+
+    BOOST_CHECK(!net::is_disconnect_exception(std::make_exception_ptr(not_a_de))
+                   .has_value());
+
+    BOOST_CHECK(
+      net::is_disconnect_exception(
+        std::make_exception_ptr(ss::nested_exception(
+          std::make_exception_ptr(is_a_de), std::make_exception_ptr(not_a_de))))
+        .has_value());
+
+    BOOST_CHECK(
+      net::is_disconnect_exception(
+        std::make_exception_ptr(ss::nested_exception(
+          std::make_exception_ptr(not_a_de), std::make_exception_ptr(is_a_de))))
+        .has_value());
+
+    BOOST_CHECK(!net::is_disconnect_exception(
+                   std::make_exception_ptr(ss::nested_exception(
+                     std::make_exception_ptr(not_a_de),
+                     std::make_exception_ptr(not_a_de))))
+                   .has_value());
+}