ct/l1: change metastore *_objects signatures to take const refs

Change add_objects, replace_objects, and commit_objects method
signatures to take const refs instead of unique_ptr by value.
This enables retry scenarios where the builder needs to remain usable
after the call.

The replicated metastore's release method is no longer used and is
removed. The simple metastore still uses this method for testing, so it
remains.

Author: wdberkeley

src/v/cloud_topics/level_one/frontend_reader/tests/l1_reader_fixture.h

@@ -125,7 +125,7 @@ class l1_reader_fixture : public seastar_test {
               .value();
         }
 
-        _metastore.add_objects(std::move(meta_builder), term_map).get().value();
+        _metastore.add_objects(*meta_builder, term_map).get().value();
     }
 
     model::record_batch_reader make_reader(

src/v/cloud_topics/level_one/frontend_reader/tests/reader_test.cc

@@ -304,7 +304,7 @@ TEST_F(l1_reader_test, missing_object) {
         .first_offset = kafka::offset{0},
       });
 
-    _metastore.add_objects(std::move(builder), term_map).get().value();
+    _metastore.add_objects(*builder, term_map).get().value();
 
     auto reader = make_reader(ntp, tidp);
     EXPECT_THROW(read_all(std::move(reader)), std::runtime_error);
@@ -364,7 +364,7 @@ TEST_F(l1_reader_test, empty_offset_range) {
         .term = model::term_id{1},
         .first_offset = high_watermark,
       });
-    _metastore.add_objects(std::move(meta_builder), term_map).get().value();
+    _metastore.add_objects(*meta_builder, term_map).get().value();
 
     // Write some objects after the empty object.
     auto final_batches

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

@@ -170,9 +170,8 @@ class metastore {
     // it does not imply that all extents were accepted by the metastore. The
     // response should be examined to determine if subsequent add_objects()
     // requests need to be re-aligned to a different offset.
-    virtual ss::future<std::expected<add_response, errc>> add_objects(
-      std::unique_ptr<object_metadata_builder>, const term_offset_map_t&)
-      = 0;
+    virtual ss::future<std::expected<add_response, errc>>
+    add_objects(const object_metadata_builder&, const term_offset_map_t&) = 0;
 
     // Adds the given objects to the metastore, expecting that the new extents
     // replace an extent or set of extents covering the same range.
@@ -183,7 +182,7 @@ class metastore {
     // correctness, these simplify accounting and makes it easier to validate
     // that we haven't lost data.
     virtual ss::future<std::expected<void, errc>>
-      replace_objects(std::unique_ptr<object_metadata_builder>) = 0;
+    replace_objects(const object_metadata_builder&) = 0;
 
     // Moves the start offset of the given partition's log to the given offset.
     virtual ss::future<std::expected<void, errc>>
@@ -277,8 +276,8 @@ class metastore {
     // Similar to replace_objects(), but with additional constraints based on
     // compaction metadata. See get_compaction_offsets() for more details on
     // expected usage.
-    virtual ss::future<std::expected<void, errc>> compact_objects(
-      std::unique_ptr<object_metadata_builder>, const compaction_map_t&)
+    virtual ss::future<std::expected<void, errc>>
+    compact_objects(const object_metadata_builder&, const compaction_map_t&)
       = 0;
 
     // Returns metadata required to determine what to compact for the given

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

@@ -102,12 +102,6 @@ class replicated_object_builder : public metastore::object_metadata_builder {
 
 private:
     friend class cloud_topics::l1::replicated_metastore;
-    std::expected<
-      chunked_hash_map<
-        model::partition_id,
-        chunked_vector<metastore::object_metadata>>,
-      error>
-    release();
 
     struct partitioned_objects {
         chunked_hash_map<
@@ -196,33 +190,6 @@ replicated_object_builder::finish(
     return {};
 }
 
-std::expected<
-  chunked_hash_map<
-    model::partition_id,
-    chunked_vector<metastore::object_metadata>>,
-  replicated_object_builder::error>
-replicated_object_builder::release() {
-    for (const auto& [partition_id, partition_objects] : partitions_) {
-        if (!partition_objects.pending_objects_.empty()) {
-            return std::unexpected(error{"Unfinished objects remain"});
-        }
-    }
-
-    chunked_hash_map<
-      model::partition_id,
-      chunked_vector<metastore::object_metadata>>
-      result;
-    for (auto& [partition_id, partition_objects] : partitions_) {
-        chunked_vector<metastore::object_metadata> metas;
-        for (auto& obj : partition_objects.finished_objects_) {
-            metas.push_back(std::move(obj));
-        }
-        result.emplace(partition_id, std::move(metas));
-    }
-
-    partitions_.clear();
-    return result;
-}
 } // anonymous namespace
 
 replicated_metastore::replicated_metastore(frontend& fe)
@@ -258,18 +225,19 @@ replicated_metastore::get_offsets(const model::topic_id_partition& tidp) {
 
 ss::future<std::expected<metastore::add_response, metastore::errc>>
 replicated_metastore::add_objects(
-  std::unique_ptr<metastore::object_metadata_builder> builder,
+  const metastore::object_metadata_builder& builder,
   const metastore::term_offset_map_t& terms) {
-    auto replicated_builder = std::unique_ptr<replicated_object_builder>(
-      static_cast<replicated_object_builder*>(builder.release()));
-
-    auto objects_result = replicated_builder->release();
-    if (!objects_result.has_value()) {
-        vlog(
-          cd_log.error,
-          "Error while sending request: {}",
-          objects_result.error());
-        co_return std::unexpected(metastore::errc::invalid_request);
+    auto& replicated_builder = static_cast<const replicated_object_builder&>(
+      builder);
+
+    for (const auto& [partition_id, partition_objects] :
+         replicated_builder.partitions_) {
+        if (!partition_objects.pending_objects_.empty()) {
+            vlog(
+              cd_log.error,
+              "Error while sending request: unfinished objects remain");
+            co_return std::unexpected(metastore::errc::invalid_request);
+        }
     }
     chunked_hash_map<model::partition_id, term_state_update_t>
       partitioned_terms;
@@ -286,7 +254,8 @@ replicated_metastore::add_objects(
         }
     }
     add_response resp;
-    for (auto& [partition_id, partition_objects] : objects_result.value()) {
+    for (auto& [partition_id, partition_objects] :
+         replicated_builder.partitions_) {
         auto terms_it = partitioned_terms.find(partition_id);
         if (terms_it == partitioned_terms.end()) {
             // TODO: consider making this less strict, down to the STM layer?
@@ -300,7 +269,7 @@ replicated_metastore::add_objects(
         req.metastore_partition = partition_id;
         req.new_terms = std::move(terms_it->second);
         chunked_vector<new_object> new_objects;
-        for (auto& obj : partition_objects) {
+        for (auto& obj : partition_objects.finished_objects_) {
             new_objects.emplace_back(meta_to_rpc_obj(obj));
         }
         req.new_objects = std::move(new_objects);
@@ -325,23 +294,25 @@ replicated_metastore::add_objects(
 
 ss::future<std::expected<void, metastore::errc>>
 replicated_metastore::replace_objects(
-  std::unique_ptr<metastore::object_metadata_builder> builder) {
-    auto replicated_builder = std::unique_ptr<replicated_object_builder>(
-      static_cast<replicated_object_builder*>(builder.release()));
-
-    auto objects_result = replicated_builder->release();
-    if (!objects_result) {
-        vlog(
-          cd_log.error,
-          "Error while sending request: {}",
-          objects_result.error());
-        co_return std::unexpected(metastore::errc::invalid_request);
+  const metastore::object_metadata_builder& builder) {
+    auto& replicated_builder = static_cast<const replicated_object_builder&>(
+      builder);
+
+    for (const auto& [partition_id, partition_objects] :
+         replicated_builder.partitions_) {
+        if (!partition_objects.pending_objects_.empty()) {
+            vlog(
+              cd_log.error,
+              "Error while sending request: unfinished objects remain");
+            co_return std::unexpected(metastore::errc::invalid_request);
+        }
     }
-    for (auto& [partition_id, partition_objects] : objects_result.value()) {
+    for (auto& [partition_id, partition_objects] :
+         replicated_builder.partitions_) {
         rpc::replace_objects_request req;
         req.metastore_partition = partition_id;
         chunked_vector<new_object> new_objects;
-        for (auto& obj : partition_objects) {
+        for (auto& obj : partition_objects.finished_objects_) {
             new_objects.emplace_back(meta_to_rpc_obj(obj));
         }
         req.new_objects = std::move(new_objects);
@@ -514,18 +485,19 @@ replicated_metastore::get_end_offset_for_term(
 
 ss::future<std::expected<void, metastore::errc>>
 replicated_metastore::compact_objects(
-  std::unique_ptr<metastore::object_metadata_builder> builder,
+  const metastore::object_metadata_builder& builder,
   const metastore::compaction_map_t& compaction_updates) {
-    auto replicated_builder = std::unique_ptr<replicated_object_builder>(
-      static_cast<replicated_object_builder*>(builder.release()));
-
-    auto objects_result = replicated_builder->release();
-    if (!objects_result) {
-        vlog(
-          cd_log.error,
-          "Error while sending request: {}",
-          objects_result.error());
-        co_return std::unexpected(metastore::errc::invalid_request);
+    auto& replicated_builder = static_cast<const replicated_object_builder&>(
+      builder);
+
+    for (const auto& [partition_id, partition_objects] :
+         replicated_builder.partitions_) {
+        if (!partition_objects.pending_objects_.empty()) {
+            vlog(
+              cd_log.error,
+              "Error while sending request: unfinished objects remain");
+            co_return std::unexpected(metastore::errc::invalid_request);
+        }
     }
     chunked_hash_map<
       model::partition_id,
@@ -537,7 +509,7 @@ replicated_metastore::compact_objects(
             vlog(cd_log.warn, "Unable to get metastore partition for {}", tp);
             co_return std::unexpected(errc::transport_error);
         }
-        if (!objects_result->contains(*metastore_partition)) {
+        if (!replicated_builder.partitions_.contains(*metastore_partition)) {
             vlog(
               cd_log.error,
               "Expected objects for partition {}",
@@ -547,11 +519,12 @@ replicated_metastore::compact_objects(
         compaction_updates_by_partition[*metastore_partition].emplace(
           tp, meta_to_rpc_compact_update(update));
     }
-    for (auto& [partition_id, partition_objects] : objects_result.value()) {
+    for (auto& [partition_id, partition_objects] :
+         replicated_builder.partitions_) {
         rpc::replace_objects_request req;
         req.metastore_partition = partition_id;
         chunked_vector<new_object> new_objects;
-        for (auto& obj : partition_objects) {
+        for (auto& obj : partition_objects.finished_objects_) {
             new_objects.emplace_back(meta_to_rpc_obj(obj));
         }
         req.new_objects = std::move(new_objects);

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

@@ -28,11 +28,10 @@ class replicated_metastore : public metastore {
     get_offsets(const model::topic_id_partition&) override;
 
     ss::future<std::expected<add_response, errc>> add_objects(
-      std::unique_ptr<object_metadata_builder>,
-      const term_offset_map_t&) override;
+      const object_metadata_builder&, const term_offset_map_t&) override;
 
     ss::future<std::expected<void, errc>>
-      replace_objects(std::unique_ptr<object_metadata_builder>) override;
+    replace_objects(const object_metadata_builder&) override;
 
     ss::future<std::expected<void, errc>>
     set_start_offset(const model::topic_id_partition&, kafka::offset) override;
@@ -53,8 +52,7 @@ class replicated_metastore : public metastore {
       const model::topic_id_partition&, kafka::offset) override;
 
     ss::future<std::expected<void, errc>> compact_objects(
-      std::unique_ptr<object_metadata_builder>,
-      const compaction_map_t&) override;
+      const object_metadata_builder&, const compaction_map_t&) override;
     ss::future<std::expected<void, errc>> compact_objects(
       const chunked_vector<object_metadata>&, const compaction_map_t&);
 

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

@@ -97,7 +97,8 @@ simple_object_builder::release() {
     if (!pending_objects_.empty()) {
         return std::unexpected(
           error{fmt::format(
-            "Builder still has {} pending object", pending_objects_.size())});
+            "Builder still has {} pending object(s)",
+            pending_objects_.size())});
     }
     return std::exchange(finished_objects_, {});
 }
@@ -148,15 +149,17 @@ simple_metastore::get_offsets(
 
 ss::future<std::expected<metastore::add_response, metastore::errc>>
 simple_metastore::add_objects(
-  std::unique_ptr<metastore::object_metadata_builder> builder,
+  const metastore::object_metadata_builder& builder,
   const term_offset_map_t& terms) {
-    auto* simple_builder = dynamic_cast<simple_object_builder*>(builder.get());
-    auto objects_res = simple_builder->release();
-    if (!objects_res.has_value()) {
-        vlog(cd_log.error, "Failed to add: {}", objects_res.error());
+    auto& simple_builder = dynamic_cast<const simple_object_builder&>(builder);
+    if (!simple_builder.pending_objects_.empty()) {
+        vlog(
+          cd_log.error,
+          "Failed to add: builder still has {} pending object(s)",
+          simple_builder.pending_objects_.size());
         co_return std::unexpected(metastore::errc::invalid_request);
     }
-    co_return co_await add_objects(objects_res.value(), terms);
+    co_return co_await add_objects(simple_builder.finished_objects_, terms);
 }
 
 ss::future<std::expected<metastore::add_response, metastore::errc>>
@@ -188,14 +191,16 @@ simple_metastore::add_objects(
 
 ss::future<std::expected<void, metastore::errc>>
 simple_metastore::replace_objects(
-  std::unique_ptr<metastore::object_metadata_builder> builder) {
-    auto* simple_builder = dynamic_cast<simple_object_builder*>(builder.get());
-    auto objects_res = simple_builder->release();
-    if (!objects_res.has_value()) {
-        vlog(cd_log.error, "Failed to replace: {}", objects_res.error());
+  const metastore::object_metadata_builder& builder) {
+    auto& simple_builder = dynamic_cast<const simple_object_builder&>(builder);
+    if (!simple_builder.pending_objects_.empty()) {
+        vlog(
+          cd_log.error,
+          "Failed to replace: builder still has {} pending object(s)",
+          simple_builder.pending_objects_.size());
         co_return std::unexpected(metastore::errc::invalid_request);
     }
-    co_return co_await replace_objects(objects_res.value());
+    co_return co_await replace_objects(simple_builder.finished_objects_);
 }
 
 ss::future<std::expected<void, metastore::errc>>
@@ -455,15 +460,18 @@ simple_metastore::compact_objects(
 
 ss::future<std::expected<void, metastore::errc>>
 simple_metastore::compact_objects(
-  std::unique_ptr<metastore::object_metadata_builder> builder,
+  const metastore::object_metadata_builder& builder,
   const compaction_map_t& compaction_metas) {
-    auto* simple_builder = dynamic_cast<simple_object_builder*>(builder.get());
-    auto objects_res = simple_builder->release();
-    if (!objects_res.has_value()) {
-        vlog(cd_log.error, "Failed to compact: {}", objects_res.error());
+    auto& simple_builder = dynamic_cast<const simple_object_builder&>(builder);
+    if (!simple_builder.pending_objects_.empty()) {
+        vlog(
+          cd_log.error,
+          "Failed to compact: builder still has {} pending object(s)",
+          simple_builder.pending_objects_.size());
         co_return std::unexpected(metastore::errc::invalid_request);
     }
-    co_return co_await compact_objects(objects_res.value(), compaction_metas);
+    co_return co_await compact_objects(
+      simple_builder.finished_objects_, compaction_metas);
 }
 
 ss::future<

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

@@ -57,13 +57,12 @@ class simple_metastore : public metastore {
     get_offsets(const model::topic_id_partition&) override;
 
     ss::future<std::expected<add_response, errc>> add_objects(
-      std::unique_ptr<object_metadata_builder>,
-      const term_offset_map_t&) override;
+      const object_metadata_builder&, const term_offset_map_t&) override;
     ss::future<std::expected<add_response, errc>> add_objects(
       const chunked_vector<object_metadata>&, const term_offset_map_t&);
 
     ss::future<std::expected<void, errc>>
-      replace_objects(std::unique_ptr<object_metadata_builder>) override;
+    replace_objects(const object_metadata_builder&) override;
     ss::future<std::expected<void, errc>>
     replace_objects(const chunked_vector<object_metadata>&);
 
@@ -86,8 +85,7 @@ class simple_metastore : public metastore {
       const model::topic_id_partition&, kafka::offset) override;
 
     ss::future<std::expected<void, errc>> compact_objects(
-      std::unique_ptr<object_metadata_builder>,
-      const compaction_map_t&) override;
+      const object_metadata_builder&, const compaction_map_t&) override;
     ss::future<std::expected<void, errc>> compact_objects(
       const chunked_vector<object_metadata>&, const compaction_map_t&);