[v24.1.x] CORE-8082 cloud_io: add missing error handling to

src/v/cloud_storage/remote.cc

@@ -863,12 +863,18 @@ remote::download_object(cloud_storage::download_request download_request) {
 
         if (resp) {
             vlog(ctxlog.debug, "Receive OK response from {}", path);
-            auto buffer
-              = co_await cloud_storage_clients::util::drain_response_stream(
-                resp.value());
-            download_request.payload.append_fragments(std::move(buffer));
-            transfer_details.on_success(_probe);
-            co_return download_result::success;
+            try {
+                auto buffer
+                  = co_await cloud_storage_clients::util::drain_response_stream(
+                    resp.value());
+                download_request.payload.append_fragments(std::move(buffer));
+                transfer_details.on_success(_probe);
+                co_return download_result::success;
+            } catch (...) {
+                resp
+                  = cloud_storage_clients::util::handle_client_transport_error(
+                    std::current_exception(), ctxlog);
+            }
         }
 
         lease.client->shutdown();

src/v/cloud_storage_clients/util.cc

@@ -12,6 +12,7 @@
 
 #include "bytes/streambuf.h"
 #include "net/connection.h"
+#include "utils/retry_chain_node.h"
 
 #include <boost/property_tree/xml_parser.hpp>
 
@@ -38,8 +39,9 @@ bool has_abort_or_gate_close_exception(const ss::nested_exception& ex) {
            || is_abort_or_gate_close_exception(ex.outer);
 }
 
+template<typename Logger>
 error_outcome handle_client_transport_error(
-  std::exception_ptr current_exception, ss::logger& logger) {
+  std::exception_ptr current_exception, Logger& logger) {
     auto outcome = error_outcome::retry;
 
     try {
@@ -112,6 +114,11 @@ error_outcome handle_client_transport_error(
     return outcome;
 }
 
+template error_outcome
+handle_client_transport_error<ss::logger>(std::exception_ptr, ss::logger&);
+template error_outcome handle_client_transport_error<retry_chain_logger>(
+  std::exception_ptr, retry_chain_logger&);
+
 ss::future<iobuf>
 drain_response_stream(http::client::response_stream_ref resp) {
     const auto transfer_encoding = resp->get_headers().find(

src/v/cloud_storage_clients/util.h

@@ -24,8 +24,9 @@ namespace cloud_storage_clients::util {
 /// cloud provider (e.g. connection error).
 /// \param current_exception is the current exception thrown by the client
 /// \param logger is the logger to use
+template<typename Logger>
 error_outcome handle_client_transport_error(
-  std::exception_ptr current_exception, ss::logger& logger);
+  std::exception_ptr current_exception, Logger& logger);
 
 /// \brief: Drain the reponse stream pointed to by the 'resp' handle into an
 /// iobuf