Merge pull request #29509 from WillemKauf/extent_metadata_fetch

`ct/l1`: add `end_of_stream` flag to `extent_metadata_response`

Author: WillemKauf

src/v/cloud_topics/level_one/domain/db_domain_manager.cc

@@ -843,11 +843,14 @@ db_domain_manager::get_extent_metadata(rpc::get_extent_metadata_request req) {
     }
     if (!extents_result->has_value()) {
         co_return rpc::get_extent_metadata_reply{
-          .ec = rpc::errc::out_of_range,
+          .ec = rpc::errc::ok,
+          .extents = {},
+          .end_of_stream = true,
         };
     }
 
     chunked_vector<rpc::extent_metadata> extents;
+    bool end_of_stream = true;
     auto gen = (*extents_result)->get_rows();
     while (auto row_opt = co_await gen()) {
         const auto& row = row_opt->get();
@@ -866,13 +869,15 @@ db_domain_manager::get_extent_metadata(rpc::get_extent_metadata_request req) {
             .max_timestamp = extent.val.max_timestamp,
           });
         if (extents.size() >= req.max_num_extents) {
+            end_of_stream = false;
             break;
         }
     }
 
     co_return rpc::get_extent_metadata_reply{
       .ec = rpc::errc::ok,
       .extents = std::move(extents),
+      .end_of_stream = end_of_stream,
     };
 }
 

src/v/cloud_topics/level_one/domain/simple_domain_manager.cc

@@ -638,7 +638,8 @@ simple_domain_manager::get_extent_metadata(
     }
     co_return rpc::get_extent_metadata_reply{
       .ec = rpc::errc::ok,
-      .extents = meta_to_rpc_extent_metadata(std::move(get_res->extents))};
+      .extents = meta_to_rpc_extent_metadata(std::move(get_res->extents)),
+      .end_of_stream = get_res->end_of_stream};
 }
 
 ss::future<rpc::flush_domain_reply>

src/v/cloud_topics/level_one/domain/tests/db_domain_manager_test.cc

@@ -777,10 +777,11 @@ TEST_F(DbDomainManagerTest, TestRestoreWithConcurrentReads) {
                      .max_num_extents = std::numeric_limits<size_t>::max()})
                   .then([&num_reads](auto reply) {
                       ++num_reads;
-                      if (reply.ec != l1_rpc::errc::out_of_range) {
-                          EXPECT_EQ(reply.ec, l1_rpc::errc::ok);
+                      if (reply.ec == l1_rpc::errc::ok) {
                           auto num_extents = reply.extents.size();
-                          EXPECT_TRUE(num_extents == 3)
+                          // Readers can see either 0 or 3 extents, depending on
+                          // when their read arrives.
+                          EXPECT_TRUE(num_extents == 0 || num_extents == 3)
                             << "Unexpected number of extents: " << num_extents;
                       }
                   });

src/v/cloud_topics/level_one/metastore/extent_metadata_reader.cc

@@ -61,16 +61,23 @@ extent_metadata_reader::forward_generator() {
         } else {
             // Yield extents.
             auto extents = std::move(extent_md_res->extents);
+            bool end_of_stream = extent_md_res->end_of_stream;
 
-            if (extents.empty()) {
-                break;
-            }
+            dassert(
+              end_of_stream || !extents.empty(),
+              "end_of_stream=false requires non-empty extents");
 
             for (const auto& extent : extents) {
                 co_yield extent;
             }
 
-            next_offset = kafka::next_offset(extents.back().last_offset);
+            if (end_of_stream) {
+                break;
+            }
+
+            if (!extents.empty()) {
+                next_offset = kafka::next_offset(extents.back().last_offset);
+            }
         }
     }
 }
@@ -94,16 +101,23 @@ extent_metadata_reader::backward_generator() {
         } else {
             // Yield extents.
             auto extents = std::move(extent_md_res->extents);
+            bool end_of_stream = extent_md_res->end_of_stream;
 
-            if (extents.empty()) {
-                break;
-            }
+            dassert(
+              end_of_stream || !extents.empty(),
+              "end_of_stream=false requires non-empty extents");
 
             for (const auto& extent : extents) {
                 co_yield extent;
             }
 
-            next_offset = kafka::prev_offset(extents.back().base_offset);
+            if (end_of_stream) {
+                break;
+            }
+
+            if (!extents.empty()) {
+                next_offset = kafka::prev_offset(extents.back().base_offset);
+            }
         }
     }
 }

src/v/cloud_topics/level_one/metastore/metastore.h

@@ -454,6 +454,10 @@ class metastore {
 
     struct extent_metadata_response {
         extent_metadata_vec extents{};
+        // True when no more extents exist beyond this response (end of
+        // iteration). False when more extents may exist (hit max_num_extents
+        // limit).
+        bool end_of_stream{true};
     };
 
     // Returns a number of extents in the offset range `[start, end]`

src/v/cloud_topics/level_one/metastore/replicated_metastore.cc

@@ -834,6 +834,7 @@ replicated_metastore::get_extent_metadata_forwards(
 
     metastore::extent_metadata_response resp;
     resp.extents = rpc_to_meta_extent_metadata(std::move(reply.extents));
+    resp.end_of_stream = reply.end_of_stream;
 
     co_return resp;
 }
@@ -869,6 +870,7 @@ replicated_metastore::get_extent_metadata_backwards(
 
     metastore::extent_metadata_response resp;
     resp.extents = rpc_to_meta_extent_metadata(std::move(reply.extents));
+    resp.end_of_stream = reply.end_of_stream;
 
     co_return resp;
 }

src/v/cloud_topics/level_one/metastore/rpc_types.h

@@ -360,10 +360,13 @@ struct get_extent_metadata_reply
       get_extent_metadata_reply,
       serde::version<0>,
       serde::compat_version<0>> {
-    auto serde_fields() { return std::tie(ec, extents); }
+    auto serde_fields() { return std::tie(ec, extents, end_of_stream); }
 
     errc ec;
     chunked_vector<extent_metadata> extents;
+    // True when no more extents exist beyond this response (end of iteration).
+    // False when more extents may exist (hit max_num_extents limit).
+    bool end_of_stream{true};
 };
 struct get_extent_metadata_request
   : serde::envelope<

src/v/cloud_topics/level_one/metastore/simple_metastore.cc

@@ -764,6 +764,7 @@ simple_metastore::get_extent_metadata_forwards(
     const auto& prt = prt_ref->get();
 
     extent_metadata_vec extents;
+    bool end_of_stream = true;
 
     auto min_it = std::ranges::lower_bound(
       prt.extents, min_offset, std::less<>{}, &extent::last_offset);
@@ -773,17 +774,19 @@ simple_metastore::get_extent_metadata_forwards(
             break;
         }
 
-        if (extents.size() >= max_num_extents) {
-            break;
-        }
-
         extents.push_back(
           {.base_offset = extent.base_offset,
            .last_offset = extent.last_offset,
            .max_timestamp = extent.max_timestamp});
+
+        if (extents.size() >= max_num_extents) {
+            end_of_stream = false;
+            break;
+        }
     }
 
-    return extent_metadata_response{.extents = std::move(extents)};
+    return extent_metadata_response{
+      .extents = std::move(extents), .end_of_stream = end_of_stream};
 }
 
 ss::future<std::expected<metastore::extent_metadata_response, metastore::errc>>
@@ -813,6 +816,7 @@ simple_metastore::get_extent_metadata_backwards(
     const auto& prt = prt_ref->get();
 
     extent_metadata_vec extents;
+    bool end_of_stream = true;
 
     auto max_it = std::ranges::lower_bound(
       prt.extents, max_offset, std::less<>{}, &extent::last_offset);
@@ -833,17 +837,19 @@ simple_metastore::get_extent_metadata_backwards(
             break;
         }
 
-        if (extents.size() >= max_num_extents) {
-            break;
-        }
-
         extents.push_back(
           {.base_offset = extent.base_offset,
            .last_offset = extent.last_offset,
            .max_timestamp = extent.max_timestamp});
+
+        if (extents.size() >= max_num_extents) {
+            end_of_stream = false;
+            break;
+        }
     }
 
-    return extent_metadata_response{.extents = std::move(extents)};
+    return extent_metadata_response{
+      .extents = std::move(extents), .end_of_stream = end_of_stream};
 }
 
 } // namespace cloud_topics::l1