ct/l1: add end_of_stream flag to extent_metadata_response

This flag serves as a strong indicator that we have iterated over all
the extents requested in an offset range, rather than having an empty
extent list represent this case.

Also change the handling of the extent response in the `db_domain_manager`
to return an empty list of extents with `end_of_stream=true` instead
of an error when no extents are found in the requested range.

The edge case of `extents.empty()` and `end_of_stream=false` potentially
resulting in an infinite loop in `extent_metadata_reader` is also handled
by ensuring that a request for `0` extents results in a single extent being
returned with `end_of_stream=false`.

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