From 57706bfefbc4c555e095302644b579943bbaf913 Mon Sep 17 00:00:00 2001
From: Abhijat Malviya <abhijat@redpanda.com>
Date: Mon, 22 May 2023 14:15:38 +0530
Subject: [PATCH] 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.
---
 .../tests/CMakeLists.txt                      |  1 +
 .../tests/exception_test.cc                   | 40 +++++++++++++++++++
 src/v/cloud_storage_clients/util.cc           | 33 +++++++++++++++
 src/v/cloud_storage_clients/util.h            |  2 +
 4 files changed, 76 insertions(+)
 create mode 100644 src/v/cloud_storage_clients/tests/exception_test.cc

diff --git a/src/v/cloud_storage_clients/tests/CMakeLists.txt b/src/v/cloud_storage_clients/tests/CMakeLists.txt
index 6e47e0e3a3fa7..88e9990147047 100644
--- a/src/v/cloud_storage_clients/tests/CMakeLists.txt
+++ b/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"
diff --git a/src/v/cloud_storage_clients/tests/exception_test.cc b/src/v/cloud_storage_clients/tests/exception_test.cc
new file mode 100644
index 0000000000000..341a4b330639a
--- /dev/null
+++ b/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));
+    }
+}
diff --git a/src/v/cloud_storage_clients/util.cc b/src/v/cloud_storage_clients/util.cc
index ce7c1d450adc1..5dd6d75c541a0 100644
--- a/src/v/cloud_storage_clients/util.cc
+++ b/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;
diff --git a/src/v/cloud_storage_clients/util.h b/src/v/cloud_storage_clients/util.h
index 1cd04ece853f0..b946b8c40d2fa 100644
--- a/src/v/cloud_storage_clients/util.h
+++ b/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