Merge pull request #29809 from redpanda-data/l0-reader-fail-mat

ct/l0: allow materialization to skip missing extents

Author: wdberkeley

src/v/cloud_topics/BUILD

@@ -47,11 +47,13 @@ redpanda_cc_library(
     ],
     visibility = [":__subpackages__"],
     deps = [
+        "//src/v/base",
         "//src/v/random:generators",
         "//src/v/serde",
         "//src/v/utils:named_type",
         "//src/v/utils:uuid",
         "@fmt",
+        "@seastar",
     ],
 )
 
@@ -257,6 +259,7 @@ redpanda_cc_library(
     ],
     visibility = ["//visibility:public"],
     deps = [
+        ":types",
         "//src/v/base",
         "//src/v/model",
         "//src/v/utils:to_string",

src/v/cloud_topics/data_plane_api.h

@@ -73,12 +73,16 @@ class data_plane_api {
 
     // Materialize extents from the L0 read pipeline.
     // `output_size_estimate` must not exceed `materialize_max_bytes()`.
+    // When `allow_mat_failure` is yes, download_not_found (404)
+    // errors for individual extents are tolerated: the missing extents are
+    // skipped and the result contains fewer batches than requested.
     virtual ss::future<result<chunked_vector<model::record_batch>>> materialize(
       model::ntp ntp,
       size_t output_size_estimate,
       chunked_vector<extent_meta> metadata,
       model::timeout_clock::time_point timeout,
-      model::opt_abort_source_t)
+      model::opt_abort_source_t,
+      allow_materialization_failure allow_mat_failure)
       = 0;
 
     /// Return the maximum bytes that may be requested in a single

src/v/cloud_topics/data_plane_impl.cc

@@ -165,7 +165,8 @@ class impl
       size_t output_size_estimate,
       chunked_vector<extent_meta> metadata,
       model::timeout_clock::time_point timeout,
-      model::opt_abort_source_t as) override {
+      model::opt_abort_source_t as,
+      allow_materialization_failure allow_mat_failure) override {
         if (metadata.empty()) {
             co_return chunked_vector<model::record_batch>{};
         }
@@ -183,6 +184,7 @@ class impl
           {
             .output_size_estimate = output_size_estimate,
             .meta = std::move(metadata),
+            .allow_mat_failure = allow_mat_failure,
           },
           timeout,
           as);

src/v/cloud_topics/frontend/tests/frontend_test.cc

@@ -63,7 +63,8 @@ class mock_api : public data_plane_api {
        size_t output_size_estimate,
        chunked_vector<extent_meta> metadata,
        model::timeout_clock::time_point timeout,
-       model::opt_abort_source_t),
+       model::opt_abort_source_t,
+       allow_materialization_failure allow_mat_failure),
       (override));
 
     MOCK_METHOD(

src/v/cloud_topics/level_zero/frontend_reader/level_zero_reader.cc

@@ -386,7 +386,8 @@ level_zero_log_reader_impl::materialize_batches(
           materialize_bytes,
           std::move(to_materialize),
           deadline,
-          _config.abort_source);
+          _config.abort_source,
+          _config.allow_mat_failure);
         if (!mat_res.has_value()) {
             if (mat_res.error() == errc::shutting_down) {
                 vlog(_log.debug, "Materialize aborted due to shutdown");
@@ -402,7 +403,10 @@ level_zero_log_reader_impl::materialize_batches(
               mat_res.error().message()));
         }
         batches = std::move(mat_res.value());
-        if (batches.size() != materialize_count) {
+        auto count_ok = bool(_config.allow_mat_failure)
+                          ? batches.size() <= materialize_count
+                          : batches.size() == materialize_count;
+        if (!count_ok) {
             throw std::runtime_error(fmt_with_ctx(
               fmt::format,
               "Materialized unexpected number of batches: {}, expected: {}",
@@ -411,7 +415,10 @@ level_zero_log_reader_impl::materialize_batches(
         }
     }
     // Merge our selected subset of unhydrated batches with the materialized
-    // batches, preserving control batches from the local log.
+    // batches, preserving control batches from the local log. When
+    // allow_mat_failure is set, some extents may have been skipped: the
+    // materialized batches are a subsequence of the query (same offset
+    // order), so sequential offset comparison identifies which were skipped.
     auto batches_it = batches.begin();
     chunked_circular_buffer<model::record_batch> hydrated;
     auto range_to_materialize = std::ranges::subrange(
@@ -421,10 +428,27 @@ level_zero_log_reader_impl::materialize_batches(
             _config.abort_source.value().get().check();
         }
         auto& local_batch_header = local_batch.header;
-        model::record_batch batch = ss::visit(
+        auto maybe_batch = ss::visit(
           local_batch.data,
-          [this, &local_batch_header, &batches_it, &tidp](
-            const cloud_topics::extent_meta&) {
+          [this, &local_batch_header, &batches_it, &batches, &tidp](
+            const cloud_topics::extent_meta& meta)
+            -> std::optional<model::record_batch> {
+              if (
+                batches_it == batches.end()
+                || batches_it->base_offset()
+                     != kafka::offset_cast(meta.base_offset)) {
+                  if (!bool(_config.allow_mat_failure)) {
+                      throw std::runtime_error(fmt_with_ctx(
+                        fmt::format,
+                        "Materialized batch offset mismatch: expected "
+                        "{}, got {}",
+                        kafka::offset_cast(meta.base_offset),
+                        batches_it == batches.end()
+                          ? model::offset{}
+                          : batches_it->base_offset()));
+                  }
+                  return std::nullopt;
+              }
               model::record_batch batch = apply_placeholder_to_batch(
                 local_batch_header, std::move(*batches_it));
               ++batches_it;
@@ -440,19 +464,23 @@ level_zero_log_reader_impl::materialize_batches(
               }
               return batch;
           },
-          [&local_batch_header](local_log_batch::payload& payload) {
+          [&local_batch_header](local_log_batch::payload& payload)
+            -> std::optional<model::record_batch> {
               return model::record_batch(
                 local_batch_header,
                 std::move(payload),
                 model::record_batch::tag_ctor_ng{});
           },
-          [](local_log_batch::cached_batch& cb) {
+          [](local_log_batch::cached_batch& cb)
+            -> std::optional<model::record_batch> {
               // Cache hit resolved during the collection loop above.
               // The batch is already fully formed (apply_placeholder_to_batch
               // was applied before cache_put on the path that populated it).
               return std::move(cb.batch);
           });
-        hydrated.push_back(std::move(batch));
+        if (maybe_batch.has_value()) {
+            hydrated.push_back(std::move(*maybe_batch));
+        }
         co_await ss::coroutine::maybe_yield();
     }
     vassert(

src/v/cloud_topics/level_zero/pipeline/read_request.h

@@ -38,6 +38,8 @@ struct dataplane_query_result {
 struct dataplane_query {
     size_t output_size_estimate{0};
     chunked_vector<extent_meta> meta;
+    allow_materialization_failure allow_mat_failure
+      = allow_materialization_failure::no;
 };
 
 // This object is created for every fetch request.

src/v/cloud_topics/level_zero/reader/fetch_request_handler.cc

@@ -105,13 +105,13 @@ ss::future<> fetch_handler::process_single_request(l0::read_request<>* req) {
     std::optional<chunked_vector<model::tx_range>> aborted_tx;
     try {
         auto meta = std::move(req->query.meta);
-
         auto extent = co_await ss::coroutine::as_future(
           materialize_placeholders(
             _bucket,
             std::move(meta),
             *_remote,
             *_cache,
+            req->query.allow_mat_failure,
             req->rtc,
             req->rtc_logger));
 

src/v/cloud_topics/level_zero/reader/materialized_extent_reader.cc

@@ -37,6 +37,7 @@ ss::future<result<chunked_vector<materialized_extent>>> materialize_sorted_run(
   cloud_storage_clients::bucket_name bucket,
   cloud_io::remote_api<>* api,
   cloud_io::basic_cache_service_api<>* cache,
+  allow_materialization_failure allow_mat_failure,
   retry_chain_node* rtc,
   micro_probe* probe) {
     absl::node_hash_map<object_id, iobuf> hydrated;
@@ -54,6 +55,19 @@ ss::future<result<chunked_vector<materialized_extent>>> materialize_sorted_run(
             auto res = co_await materialize(
               &back, bucket, api, cache, rtc, probe);
             if (!res.has_value()) {
+                if (
+                  bool(allow_mat_failure)
+                  && res.error() == errc::download_not_found) {
+                    vlog(
+                      cd_log.warn,
+                      "Skipping extent {} (offsets {}~{}) due to missing "
+                      "object",
+                      back.meta.id,
+                      back.meta.base_offset,
+                      back.meta.last_offset);
+                    extents.pop_back();
+                    continue;
+                }
                 co_return res.error();
             }
             // If reading from cache (res.value() == true), only the
@@ -78,11 +92,12 @@ ss::future<materialize_result> materialize_placeholders(
   chunked_vector<extent_meta> query,
   cloud_io::remote_api<ss::lowres_clock>& api,
   cloud_io::basic_cache_service_api<ss::lowres_clock>& cache,
+  allow_materialization_failure allow_mat_failure,
   retry_chain_node& rtc,
   retry_chain_logger& logger) {
     micro_probe probe;
     auto extents = co_await materialize_sorted_run(
-      std::move(query), bucket, &api, &cache, &rtc, &probe);
+      std::move(query), bucket, &api, &cache, allow_mat_failure, &rtc, &probe);
     if (!extents.has_value()) {
         vlog(
           logger.warn,

src/v/cloud_topics/level_zero/reader/materialized_extent_reader.h

@@ -33,18 +33,20 @@ struct materialize_result {
 /// The method processes 'underlying' fully. The result is stored
 /// in memory so the caller should be careful with this param.
 ///
-/// \param cfg is a log reader config
 /// \param bucket is a cloud storage bucket
 /// \param query is an array of extent_meta objects
 /// \param api is a cloud_io::remote instance
 /// \param cache is a cloud storage cache instance
+/// \param allow_mat_failure when yes, 404 errors for individual extents are
+///        tolerated and those extents are skipped
 /// \param rtc is a retry chain node to use
 /// \param rtc_logger is a logger that should track the progress
 ss::future<materialize_result> materialize_placeholders(
   cloud_storage_clients::bucket_name bucket,
   chunked_vector<extent_meta> query,
   cloud_io::remote_api<ss::lowres_clock>& api,
   cloud_io::basic_cache_service_api<ss::lowres_clock>& cache,
+  allow_materialization_failure allow_mat_failure,
   retry_chain_node& rtc,
   retry_chain_logger& logger);
 

src/v/cloud_topics/level_zero/reader/tests/materialized_extent_reader_test.cc

@@ -54,6 +54,7 @@ TEST_F_CORO(materialized_extent_fixture, full_scan_test) {
       std::move(underlying),
       remote,
       cache,
+      cloud_topics::allow_materialization_failure::no,
       rtc,
       logger);
     ASSERT_EQ_CORO(actual.value().size(), expected.size());
@@ -103,6 +104,7 @@ ss::future<> test_aggregated_log_partial_scan(
       std::move(underlying),
       fx->remote,
       fx->cache,
+      cloud_topics::allow_materialization_failure::no,
       rtc,
       logger);
 
@@ -134,9 +136,123 @@ TEST_F_CORO(materialized_extent_fixture, timeout_test) {
       std::move(underlying),
       remote,
       cache,
+      cloud_topics::allow_materialization_failure::no,
       rtc,
       logger);
 
     ASSERT_TRUE_CORO(!actual.has_value());
     ASSERT_TRUE_CORO(actual.error() == cloud_topics::errc::timeout);
 }
+
+TEST_F_CORO(
+  materialized_extent_fixture, allow_materialization_failure_skips_notfound) {
+    // When allow_materialization_failure is set and an extent's object returns
+    // 404, that extent should be skipped and the result should be empty.
+    co_await add_random_batches(1);
+
+    std::queue<injected_failure> failures;
+    failures.push({.cloud_get = injected_cloud_get_failure::return_notfound});
+    produce_placeholders(false, 1, std::move(failures));
+
+    auto underlying = convert_placeholders(make_underlying());
+    ss::abort_source as;
+    retry_chain_node rtc(as, 1s, 100ms);
+    retry_chain_logger logger(test_log, rtc, "materialized_extent_reader_test");
+
+    auto [actual, probe] = co_await cloud_topics::l0::materialize_placeholders(
+      cloud_storage_clients::bucket_name("test-bucket-name"),
+      std::move(underlying),
+      remote,
+      cache,
+      cloud_topics::allow_materialization_failure::yes,
+      rtc,
+      logger);
+
+    ASSERT_TRUE_CORO(actual.has_value());
+    ASSERT_EQ_CORO(actual.value().size(), 0);
+}
+
+TEST_F_CORO(
+  materialized_extent_fixture, notfound_propagates_without_failure_flag) {
+    // Without allow_materialization_failure, a 404 should propagate as an
+    // error, same as any other download failure.
+    co_await add_random_batches(1);
+
+    std::queue<injected_failure> failures;
+    failures.push({.cloud_get = injected_cloud_get_failure::return_notfound});
+    produce_placeholders(false, 1, std::move(failures));
+
+    auto underlying = convert_placeholders(make_underlying());
+    ss::abort_source as;
+    retry_chain_node rtc(as, 1s, 100ms);
+    retry_chain_logger logger(test_log, rtc, "materialized_extent_reader_test");
+
+    auto [actual, probe] = co_await cloud_topics::l0::materialize_placeholders(
+      cloud_storage_clients::bucket_name("test-bucket-name"),
+      std::move(underlying),
+      remote,
+      cache,
+      cloud_topics::allow_materialization_failure::no,
+      rtc,
+      logger);
+
+    ASSERT_TRUE_CORO(!actual.has_value());
+    ASSERT_TRUE_CORO(actual.error() == cloud_topics::errc::download_not_found);
+}
+
+TEST_F_CORO(
+  materialized_extent_fixture, non_404_error_propagates_with_failure_flag) {
+    // Only 404 errors are tolerated. A timeout should still propagate
+    // even when allow_materialization_failure is set.
+    co_await add_random_batches(1);
+
+    std::queue<injected_failure> failures;
+    failures.push({.cloud_get = injected_cloud_get_failure::return_timeout});
+    produce_placeholders(false, 1, std::move(failures));
+
+    auto underlying = convert_placeholders(make_underlying());
+    ss::abort_source as;
+    retry_chain_node rtc(as, 1s, 100ms);
+    retry_chain_logger logger(test_log, rtc, "materialized_extent_reader_test");
+
+    auto [actual, probe] = co_await cloud_topics::l0::materialize_placeholders(
+      cloud_storage_clients::bucket_name("test-bucket-name"),
+      std::move(underlying),
+      remote,
+      cache,
+      cloud_topics::allow_materialization_failure::yes,
+      rtc,
+      logger);
+
+    ASSERT_TRUE_CORO(!actual.has_value());
+    ASSERT_TRUE_CORO(actual.error() == cloud_topics::errc::timeout);
+}
+
+TEST_F_CORO(materialized_extent_fixture, skip_one_missing_extent_among_many) {
+    // With multiple extents, only the missing one should be skipped.
+    // The other extents should materialize normally.
+    co_await add_random_batches(3);
+
+    // Single failure in the queue: exactly one of the three L0 objects
+    // will return notfound, the other two will succeed.
+    std::queue<injected_failure> failures;
+    failures.push({.cloud_get = injected_cloud_get_failure::return_notfound});
+    produce_placeholders(false, 1, std::move(failures));
+
+    auto underlying = convert_placeholders(make_underlying());
+    ss::abort_source as;
+    retry_chain_node rtc(as, 1s, 100ms);
+    retry_chain_logger logger(test_log, rtc, "materialized_extent_reader_test");
+
+    auto [actual, probe] = co_await cloud_topics::l0::materialize_placeholders(
+      cloud_storage_clients::bucket_name("test-bucket-name"),
+      std::move(underlying),
+      remote,
+      cache,
+      cloud_topics::allow_materialization_failure::yes,
+      rtc,
+      logger);
+
+    ASSERT_TRUE_CORO(actual.has_value());
+    ASSERT_EQ_CORO(actual.value().size(), 2);
+}