ct/l1: Handle exceptions better

This change improves the handling of exceptions in the L1 reader in the
following ways:
1. It removes try-catch blocks in favor of ss::coroutine::as_future.
2. It ensures that object readers are closed even in exceptional
   situations (using #1).

Author: wdberkeley

src/v/cloud_topics/level_one/frontend_reader/BUILD

@@ -20,5 +20,6 @@ redpanda_cc_library(
         "//src/v/config",
         "//src/v/model",
         "//src/v/model:timeout_clock",
+        "@seastar",
     ],
 )

src/v/cloud_topics/level_one/frontend_reader/reader.cc

@@ -14,6 +14,8 @@
 #include "config/configuration.h"
 #include "model/timeout_clock.h"
 
+#include <seastar/coroutine/as_future.hh>
+
 #include <exception>
 #include <iterator>
 #include <utility>
@@ -90,64 +92,67 @@ ss::future<> level_one_log_reader_impl::fetch_metadata(
         co_return;
     }
 
-    try {
-        auto response = co_await _metastore->get_first_ge(
-          _tidp, _config.start_offset);
-
-        if (response.has_value()) {
-            auto& obj = response.value();
-
-            vlog(
-              cd_log.debug,
-              "Found L1 object {} for {} ({}) at offset {}",
-              obj.oid,
-              _ntp,
-              _tidp,
-              _config.start_offset);
+    auto response_fut = co_await ss::coroutine::as_future(
+      _metastore->get_first_ge(_tidp, _config.start_offset));
+    if (response_fut.failed()) {
+        vlog(
+          cd_log.error,
+          "Exception fetching metadata from metastore for {} ({}): {}",
+          _ntp,
+          _tidp,
+          response_fut.get_exception());
+        _state = state::end_of_stream;
+        co_return;
+    }
 
-            auto footer_result = co_await read_footer(
-              obj.oid, obj.footer_pos, obj.object_size);
-            if (!footer_result.has_value()) {
-                vlog(
-                  cd_log.warn,
-                  "Failed to read footer from L1 object {} for {} ({}): {}",
-                  obj.oid,
-                  _ntp,
-                  _tidp,
-                  std::to_underlying(footer_result.error()));
-                _state = state::end_of_stream;
-                co_return;
-            }
-            _current_obj = current_object{
-              .oid = obj.oid, .footer = std::move(footer_result.value())};
-            _state = state::ready;
-        } else if (response.error() == l1::metastore::errc::out_of_range) {
+    auto response = response_fut.get();
+    if (!response.has_value()) {
+        if (response.error() == l1::metastore::errc::out_of_range) {
             vlog(
               cd_log.debug,
               "No L1 objects found for {} ({}) at offset {} or later",
               _ntp,
               _tidp,
               _config.start_offset);
             _state = state::end_of_stream;
-        } else {
-            vlog(
-              cd_log.warn,
-              "Failed to query metastore for {} ({}) offset {}: {}",
-              _ntp,
-              _tidp,
-              _config.start_offset,
-              response.error());
-            _state = state::end_of_stream;
+            co_return;
         }
-    } catch (...) {
         vlog(
-          cd_log.error,
-          "Failed to fetch metadata from metastore for {} ({}): {}",
+          cd_log.warn,
+          "Failed to query metastore for {} ({}) offset {}: {}",
+          _ntp,
+          _tidp,
+          _config.start_offset,
+          response.error());
+        _state = state::end_of_stream;
+        co_return;
+    }
+
+    auto& obj = response.value();
+    vlog(
+      cd_log.debug,
+      "Found L1 object {} for {} ({}) at offset {}",
+      obj.oid,
+      _ntp,
+      _tidp,
+      _config.start_offset);
+
+    auto footer_result = co_await read_footer(
+      obj.oid, obj.footer_pos, obj.object_size);
+    if (!footer_result.has_value()) {
+        vlog(
+          cd_log.warn,
+          "Failed to read footer from L1 object {} for {} ({}): {}",
+          obj.oid,
           _ntp,
           _tidp,
-          std::current_exception());
+          std::to_underlying(footer_result.error()));
         _state = state::end_of_stream;
+        co_return;
     }
+    _current_obj = current_object{
+      .oid = obj.oid, .footer = std::move(footer_result.value())};
+    _state = state::ready;
 }
 
 ss::future<std::expected<l1::footer, l1::io::errc>>
@@ -165,13 +170,41 @@ level_one_log_reader_impl::read_footer(
     auto* abort_source = _config.abort_source
                            ? &_config.abort_source.value().get()
                            : &default_abort_source;
-    auto stream_result = co_await _io->read_object(extent, abort_source);
+    auto stream_fut = co_await ss::coroutine::as_future(
+      _io->read_object(extent, abort_source));
+    if (stream_fut.failed()) {
+        vlog(
+          cd_log.error,
+          "Exception opening stream for footer from object {} (pos {} object "
+          "size {}): {}",
+          oid,
+          extent.position,
+          object_size,
+          stream_fut.get_exception());
+        co_return std::unexpected(l1::io::errc::file_io_error);
+    }
+    auto stream_result = stream_fut.get();
     if (!stream_result.has_value()) {
         co_return std::unexpected(stream_result.error());
     }
 
     auto& stream = stream_result.value();
-    iobuf footer_buf = co_await read_iobuf_exactly(stream, footer_total_size);
+
+    auto footer_fut = co_await ss::coroutine::as_future(
+      read_iobuf_exactly(stream, footer_total_size));
+    if (footer_fut.failed()) {
+        vlog(
+          cd_log.error,
+          "Exception reading footer from object {} (pos {} object size {}): {}",
+          oid,
+          extent.position,
+          object_size,
+          footer_fut.get_exception());
+        co_await stream.close();
+        co_return std::unexpected(l1::io::errc::file_io_error);
+    }
+
+    iobuf footer_buf = footer_fut.get();
     co_await stream.close();
 
     // Parse the footer - we have the complete footer so this should succeed.
@@ -213,112 +246,121 @@ ss::future<> level_one_log_reader_impl::materialize_batches(
       _current_obj.has_value(),
       "Expected to have current object in ready state");
 
-    try {
-        size_t start_pos
-          = _current_obj->footer.file_position_before_kafka_offset(
-            _tidp, _config.start_offset);
-        if (start_pos == l1::footer::npos) {
-            // This shouldn't happen- the L1 metastore told us there's
-            // data for this NTP, and we checked after the latest consume
-            // that the object isn't exhausted of data.
-            vlog(
-              cd_log.error,
-              "Reader in ready state but no data for {} ({}) in object {}",
-              _ntp,
-              _tidp,
-              _current_obj->oid);
-            _state = state::end_of_stream;
-            co_return;
-        }
+    size_t start_pos = _current_obj->footer.file_position_before_kafka_offset(
+      _tidp, _config.start_offset);
+    if (start_pos == l1::footer::npos) {
+        // This shouldn't happen- the L1 metastore told us there's
+        // data for this NTP, and we checked after the latest consume
+        // that the object isn't exhausted of data.
+        vlog(
+          cd_log.error,
+          "Reader in ready state but no data for {} ({}) in object {}",
+          _ntp,
+          _tidp,
+          _current_obj->oid);
+        _state = state::end_of_stream;
+        co_return;
+    }
 
-        l1::object_extent extent{
-          .id = _current_obj->oid,
-          .position = start_pos,
-          .size = L1_max_bytes_per_object_fetch,
-        };
-        ss::abort_source default_abort_source;
-        auto* abort_source = _config.abort_source
-                               ? &_config.abort_source.value().get()
-                               : &default_abort_source;
-        auto stream_result = co_await _io->read_object(extent, abort_source);
-        if (!stream_result.has_value()) {
+    l1::object_extent extent{
+      .id = _current_obj->oid,
+      .position = start_pos,
+      .size = L1_max_bytes_per_object_fetch,
+    };
+    ss::abort_source default_abort_source;
+    auto* abort_source = _config.abort_source
+                           ? &_config.abort_source.value().get()
+                           : &default_abort_source;
+    auto stream_fut = co_await ss::coroutine::as_future(
+      _io->read_object(extent, abort_source));
+    if (stream_fut.failed()) {
+        vlog(
+          cd_log.error,
+          "Exception opening stream for L1 object {} for {} ({}): {}",
+          _current_obj->oid,
+          _ntp,
+          _tidp,
+          stream_fut.get_exception());
+        _state = state::end_of_stream;
+        co_return;
+    }
+    auto stream_result = stream_fut.get();
+    if (!stream_result.has_value()) {
+        vlog(
+          cd_log.warn,
+          "Failed to open stream for L1 object {} for {} ({}): {}",
+          _current_obj->oid,
+          _ntp,
+          _tidp,
+          std::to_underlying(stream_result.error()));
+        _state = state::end_of_stream;
+        co_return;
+    }
+
+    auto reader = l1::object_reader::create(std::move(stream_result).value());
+    while (true) {
+        auto result_fut = co_await ss::coroutine::as_future(
+          reader->read_next());
+        if (result_fut.failed()) {
             vlog(
-              cd_log.warn,
-              "Failed to open stream for L1 object {} for {} ({}): {}",
+              cd_log.error,
+              "Exception reading L1 object {} for {} ({}): {}",
               _current_obj->oid,
               _ntp,
               _tidp,
-              std::to_underlying(stream_result.error()));
+              result_fut.get_exception());
+            co_await reader->close();
             _state = state::end_of_stream;
             co_return;
         }
 
-        auto reader = l1::object_reader::create(
-          std::move(stream_result).value());
-        while (true) {
-            auto result = co_await reader->read_next();
-
-            if (std::holds_alternative<model::record_batch>(result)) {
-                auto batch = std::move(std::get<model::record_batch>(result));
-
-                // Skip batches before our start offset.
-                if (
-                  batch.last_offset()
-                  < kafka::offset_cast(_config.start_offset)) {
-                    continue;
-                }
-
-                // Stop if we've reached our max offset.
-                if (
-                  batch.base_offset()
-                  >= kafka::offset_cast(_config.max_offset)) {
-                    break;
-                }
-
-                // Check if adding this batch would exceed our byte limit.
-                size_t batch_size = batch.size_bytes();
-                if (
-                  (_config.strict_max_bytes || _config.bytes_consumed > 0)
-                  && _config.bytes_consumed + batch_size >= _config.max_bytes) {
-                    _config.over_budget = true;
-                    break;
-                }
-                _config.bytes_consumed += batch_size;
-
-                // If we make it past all that, emit the batch.
-                _batches.push_back(std::move(batch));
-            } else if (std::holds_alternative<model::topic_id_partition>(
-                         result)) {
-                // Partition marker. Done with this ntp's partition.
+        auto result = result_fut.get();
+        if (std::holds_alternative<model::record_batch>(result)) {
+            auto batch = std::move(std::get<model::record_batch>(result));
+
+            // Skip batches before our start offset.
+            if (
+              batch.last_offset() < kafka::offset_cast(_config.start_offset)) {
+                continue;
+            }
+
+            // Stop if we've reached our max offset.
+            if (batch.base_offset() >= kafka::offset_cast(_config.max_offset)) {
                 break;
-            } else if (
-              std::holds_alternative<l1::footer>(result)
-              || std::holds_alternative<l1::object_reader::eof>(result)) {
-                // End of data.
+            }
+
+            // Check if adding this batch would exceed our byte limit.
+            size_t batch_size = batch.size_bytes();
+            if (
+              (_config.strict_max_bytes || _config.bytes_consumed > 0)
+              && _config.bytes_consumed + batch_size >= _config.max_bytes) {
+                _config.over_budget = true;
                 break;
             }
+            _config.bytes_consumed += batch_size;
+
+            // If we make it past all that, emit the batch.
+            _batches.push_back(std::move(batch));
+        } else if (std::holds_alternative<model::topic_id_partition>(result)) {
+            // Partition marker. Done with this ntp's partition.
+            break;
+        } else if (
+          std::holds_alternative<l1::footer>(result)
+          || std::holds_alternative<l1::object_reader::eof>(result)) {
+            // End of data.
+            break;
         }
+    }
 
-        co_await reader->close();
+    co_await reader->close();
 
-        vlog(
-          cd_log.debug,
-          "Materialized {} batches from L1 object {} for {} ({})",
-          _batches.size(),
-          _current_obj->oid,
-          _ntp,
-          _tidp);
-    } catch (...) {
-        vlog(
-          cd_log.error,
-          "Failed to materialize batches from L1 object {} for {} ({}): {}",
-          _current_obj->oid,
-          _ntp,
-          _tidp,
-          std::current_exception());
-        _state = state::end_of_stream;
-        co_return;
-    }
+    vlog(
+      cd_log.debug,
+      "Materialized {} batches from L1 object {} for {} ({})",
+      _batches.size(),
+      _current_obj->oid,
+      _ntp,
+      _tidp);
 
     _state = _batches.empty() ? state::end_of_stream : state::materialized;
 }