cloud_storage: Handle nested benign exception

If a nested exception is thrown where either the inner or outer
exception is abort requested or gate closed, this is logged as a debug
level error.

If the nested exception is of any other kind, an error level log is
generated.

src/v/cloud_storage_clients/tests/CMakeLists.txt

@@ -6,6 +6,7 @@ rp_test(
     backend_detection_test.cc
     s3_client_test.cc
     xml_sax_parser_test.cc
+    exception_test.cc
   DEFINITIONS BOOST_TEST_DYN_LINK
   LIBRARIES v::seastar_testing_main Boost::unit_test_framework v::http v::cloud_storage_clients v::cloud_roles
   ARGS "-- -c 1"

src/v/cloud_storage_clients/tests/exception_test.cc

@@ -0,0 +1,40 @@
+/*
+ * 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 "cloud_storage_clients/util.h"
+
+#include <boost/test/unit_test.hpp>
+
+BOOST_AUTO_TEST_CASE(test_nested_exception) {
+    {
+        auto inner = std::make_exception_ptr(ss::abort_requested_exception{});
+        auto outer = std::make_exception_ptr(ss::gate_closed_exception{});
+        ss::nested_exception ex{inner, outer};
+        BOOST_REQUIRE(
+          cloud_storage_clients::util::has_abort_or_gate_close_exception(ex));
+    }
+
+    {
+        auto inner = std::make_exception_ptr(std::bad_alloc{});
+        auto outer = std::make_exception_ptr(ss::gate_closed_exception{});
+        ss::nested_exception ex{inner, outer};
+        BOOST_REQUIRE(
+          cloud_storage_clients::util::has_abort_or_gate_close_exception(ex));
+    }
+
+    {
+        auto inner = std::make_exception_ptr(std::invalid_argument{""});
+        auto outer = std::make_exception_ptr(
+          ss::no_sharded_instance_exception{});
+        ss::nested_exception ex{inner, outer};
+        BOOST_REQUIRE(
+          !cloud_storage_clients::util::has_abort_or_gate_close_exception(ex));
+    }
+}

src/v/cloud_storage_clients/util.cc

@@ -15,8 +15,33 @@
 
 #include <boost/property_tree/xml_parser.hpp>
 
+#include <regex>
+
+namespace {
+
+bool is_abort_or_gate_close_exception(const std::exception_ptr& ex) {
+    try {
+        rethrow_exception(ex);
+    } catch (const ss::abort_requested_exception&) {
+        return true;
+    } catch (const ss::gate_closed_exception&) {
+        return true;
+    } catch (...) {
+        return false;
+    }
+}
+
+} // namespace
+
 namespace cloud_storage_clients::util {
 
+bool has_abort_or_gate_close_exception(const ss::nested_exception& ex) {
+    const auto& inner = ex.inner;
+    const auto& outer = ex.outer;
+    return is_abort_or_gate_close_exception(inner)
+           || is_abort_or_gate_close_exception(outer);
+}
+
 error_outcome handle_client_transport_error(
   std::exception_ptr current_exception, ss::logger& logger) {
     auto outcome = error_outcome::retry;
@@ -67,6 +92,14 @@ error_outcome handle_client_transport_error(
     } catch (const ss::abort_requested_exception&) {
         vlog(logger.debug, "Abort requested");
         throw;
+    } catch (const ss::nested_exception& ex) {
+        if (has_abort_or_gate_close_exception(ex)) {
+            vlog(logger.debug, "Nested abort or gate closed: {}", ex);
+            throw;
+        }
+
+        vlog(logger.error, "Unexpected error {}", std::current_exception());
+        outcome = error_outcome::fail;
     } catch (...) {
         vlog(logger.error, "Unexpected error {}", std::current_exception());
         outcome = error_outcome::fail;

src/v/cloud_storage_clients/util.h

@@ -45,4 +45,6 @@ std::chrono::system_clock::time_point parse_timestamp(std::string_view sv);
 void log_buffer_with_rate_limiting(
   const char* msg, iobuf& buf, ss::logger& logger);
 
+bool has_abort_or_gate_close_exception(const ss::nested_exception& ex);
+
 } // namespace cloud_storage_clients::util