cloud_topics/l1: delete an imported segment's tiered-storage objects

When an imported L1 object is removed, delete its backing tiered-storage objects
(the segment, its .tx range manifest, and its .index -- the same set the
archiver would have deleted) so nothing is orphaned.

- delete_objects (abstract_io/file_io/fake_io) takes object_location instead of
  object_id, so it routes by the imported location: present -> delete from the
  tiered-storage bucket; absent -> delete from the L1 bucket as before.
- Both garbage collectors (metastore and lsm) build the removal list from each
  object's imported_ts_location. Native L1 deletion is unchanged.

Testing: an imported-segment GC test (import via add_objects on a migrating
partition, then trim + GC) asserts the backing TS segment is removed, routed by
ts_path; fake_io grows has_ts_segment to observe it.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: sjust-redpanda

src/v/cloud_topics/level_one/common/abstract_io.h

@@ -25,6 +25,11 @@
 
 namespace cloud_topics::l1 {
 
+// Forward declaration — full definition in object_handle.h.
+// Do not #include object_handle.h here: object_handle.h already includes
+// abstract_io.h for io::errc, so including it here creates a cycle.
+class object_handle;
+
 // An abstraction for a local file that is used for staging uploads to object
 // storage.
 class staging_file {
@@ -51,11 +56,6 @@ class staging_file {
     virtual ss::future<ss::input_stream<char>> input_stream() = 0;
 };
 
-// Forward declaration -- full definition in object_handle.h.
-// Do not #include object_handle.h here: object_handle.h already includes this
-// header (for io::errc), so the dependency must run one way.
-class object_handle;
-
 // An abstraction for IO in level one.
 class io {
 public:
@@ -98,14 +98,21 @@ class io {
     virtual ss::future<std::expected<iobuf, errc>> read_object_as_iobuf(
       object_extent, ss::abort_source*, cloud_io::group_id g);
 
-    // Open an object for reading: returns a handle that exposes the object's
-    // index (seek by offset/timestamp) and opens readers at a seek point.
+    // Open an object for indexed read access, returning a handle that exposes
+    // both an index (for seeking) and readers (for streaming batches).
+    //
+    // For native L1 objects (extent.imported == nullopt), the handle reads the
+    // L1 footer and implements the native seek path. For imported TS extents
+    // (extent.imported.has_value()), the handle loads the TS segment index and
+    // translates log offsets to Kafka offsets while reading.
     virtual ss::future<std::expected<std::unique_ptr<object_handle>, errc>>
     open_object(object_extent, ss::abort_source*, cloud_io::group_id g) = 0;
 
-    // Delete the specified objects from object storage.
+    // Delete the specified objects from object storage. An entry with a ts_path
+    // is addressed by that tiered-storage segment path; otherwise the native L1
+    // object path is used. Both live in the one configured object bucket.
     virtual ss::future<std::expected<void, errc>>
-    delete_objects(chunked_vector<object_id>, ss::abort_source*) = 0;
+    delete_objects(chunked_vector<object_location>, ss::abort_source*) = 0;
 
     // Create a multipart upload for streaming data directly to object storage.
     virtual ss::future<

src/v/cloud_topics/level_one/common/fake_io.cc

@@ -195,14 +195,24 @@ fake_io::open_object(
       extent.id, std::get<l1::footer>(std::move(footer_result)), this, g);
 }
 
-ss::future<std::expected<void, io::errc>>
-fake_io::delete_objects(chunked_vector<object_id> oids, ss::abort_source*) {
-    for (const auto& oid : oids) {
-        remove_object(oid);
+ss::future<std::expected<void, io::errc>> fake_io::delete_objects(
+  chunked_vector<object_location> objects, ss::abort_source*) {
+    for (const auto& obj : objects) {
+        if (obj.ts_path.has_value()) {
+            // Imported object: its backing segment is addressed by ts_path
+            // (mirrors file_io's path routing).
+            _ts_storage.erase(*obj.ts_path);
+        } else {
+            remove_object(obj.id);
+        }
     }
     co_return std::expected<void, io::errc>{};
 }
 
+bool fake_io::has_ts_segment(const ts_segment_path& ts_path) const {
+    return _ts_storage.contains(ts_path);
+}
+
 std::optional<iobuf> fake_io::get_object(object_id id) {
     auto it = _storage.find(id);
     if (it == _storage.end()) {

src/v/cloud_topics/level_one/common/fake_io.h

@@ -36,7 +36,7 @@ class fake_io : public io {
       object_extent, ss::abort_source*, cloud_io::group_id g) override;
 
     ss::future<std::expected<void, errc>>
-    delete_objects(chunked_vector<object_id>, ss::abort_source*) override;
+    delete_objects(chunked_vector<object_location>, ss::abort_source*) override;
 
     ss::future<std::expected<cloud_storage_clients::multipart_upload_ref, errc>>
     create_multipart_upload(
@@ -54,6 +54,10 @@ class fake_io : public io {
     // Return a list of the object IDs that haven't been removed.
     chunked_vector<object_id> list_objects() const;
 
+    // Whether an injected TS segment (see put_ts_segment) is still present.
+    // For tests that exercise imported-object deletion.
+    bool has_ts_segment(const ts_segment_path& ts_path) const;
+
     /// Inject a raw TS-format segment for use with open_object on imported
     /// extents whose ts_path matches. open_object always seeks through the real
     /// ts_segment_index: with index_bytes (a serialized offset_index, as

src/v/cloud_topics/level_one/common/file_io.cc

@@ -441,19 +441,14 @@ file_io::open_object(
       extent.id, std::get<footer>(std::move(footer_result)), this, g);
 }
 
-ss::future<std::expected<void, io::errc>>
-file_io::delete_objects(chunked_vector<object_id> ids, ss::abort_source* as) {
-    static constexpr auto timeout = 10s;
-    static constexpr auto backoff = 100ms;
-    retry_chain_node root(*as, ss::lowres_clock::now() + timeout, backoff);
-    chunked_vector<cloud_storage_clients::object_key> keys;
-    for (const auto& id : ids) {
-        keys.push_back(object_path_factory::level_one_path(id));
-    }
+ss::future<std::expected<void, io::errc>> file_io::delete_keys(
+  const cloud_storage_clients::bucket_name& bucket,
+  chunked_vector<cloud_storage_clients::object_key> keys,
+  retry_chain_node& root) {
     auto result_fut
       = co_await ss::coroutine::as_future<cloud_io::upload_result>(
         _remote->delete_objects(
-          _bucket, std::move(keys), root, [](size_t retry_count) {
+          bucket, std::move(keys), root, [](size_t retry_count) {
               std::ignore = retry_count;
           }));
     if (result_fut.failed()) {
@@ -473,6 +468,64 @@ file_io::delete_objects(chunked_vector<object_id> ids, ss::abort_source* as) {
     std::unreachable();
 }
 
+ss::future<std::expected<void, io::errc>> file_io::delete_objects(
+  chunked_vector<object_location> objects, ss::abort_source* as) {
+    static constexpr auto timeout = 10s;
+    static constexpr auto backoff = 100ms;
+    retry_chain_node root(*as, ss::lowres_clock::now() + timeout, backoff);
+
+    chunked_vector<cloud_storage_clients::object_key> native_keys;
+    chunked_vector<cloud_storage_clients::object_key> ts_keys;
+    size_t ts_count = 0;
+    for (const auto& obj : objects) {
+        if (obj.ts_path.has_value()) {
+            ++ts_count;
+            // An imported segment owns three cloud objects (the segment, its
+            // .tx range manifest, and its .index), the same set the archiver
+            // would have deleted. Remove all three so nothing is orphaned.
+            cloud_storage::remote_segment_path seg_path{
+              std::filesystem::path{(*obj.ts_path)()}};
+            ts_keys.push_back(
+              cloud_storage_clients::object_key{(*obj.ts_path)()});
+            ts_keys.push_back(
+              cloud_storage_clients::object_key{
+                cloud_storage::generate_remote_tx_path(seg_path)().native()});
+            ts_keys.push_back(
+              cloud_storage_clients::object_key{
+                cloud_storage::generate_index_path(seg_path).native()});
+        } else {
+            native_keys.push_back(object_path_factory::level_one_path(obj.id));
+        }
+    }
+
+    if (!native_keys.empty()) {
+        auto l1_object_count = native_keys.size();
+        auto res = co_await delete_keys(_bucket, std::move(native_keys), root);
+        if (!res.has_value()) {
+            vlog(
+              cd_log.warn,
+              "Failed to delete {} L1 objects: {}",
+              l1_object_count,
+              res.error());
+            co_return res;
+        }
+    }
+    if (!ts_keys.empty()) {
+        auto ts_key_count = ts_keys.size();
+        auto res = co_await delete_keys(_bucket, std::move(ts_keys), root);
+        if (!res.has_value()) {
+            vlog(
+              cd_log.warn,
+              "Failed to delete {} keys for {} imported TS segments: {}",
+              ts_key_count,
+              ts_count,
+              res.error());
+            co_return res;
+        }
+    }
+    co_return std::expected<void, io::errc>{};
+}
+
 ss::future<std::expected<cloud_storage_clients::multipart_upload_ref, io::errc>>
 file_io::create_multipart_upload(
   object_id oid, size_t part_size, ss::abort_source* as) {

src/v/cloud_topics/level_one/common/file_io.h

@@ -48,13 +48,19 @@ class file_io : public io {
       object_extent, ss::abort_source*, cloud_io::group_id g) override;
 
     ss::future<std::expected<void, errc>>
-    delete_objects(chunked_vector<object_id>, ss::abort_source*) override;
+    delete_objects(chunked_vector<object_location>, ss::abort_source*) override;
 
     ss::future<std::expected<cloud_storage_clients::multipart_upload_ref, errc>>
     create_multipart_upload(
       object_id, size_t part_size, ss::abort_source*) override;
 
 private:
+    // Delete a batch of keys from the given bucket.
+    ss::future<std::expected<void, errc>> delete_keys(
+      const cloud_storage_clients::bucket_name& bucket,
+      chunked_vector<cloud_storage_clients::object_key> keys,
+      retry_chain_node& parent);
+
     cloud_io::remote* _remote;
     // Holds both native L1 objects and imported tiered-storage segments: a
     // single cluster always stores both in the one configured object bucket.

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

@@ -31,7 +31,7 @@ garbage_collector::remove_unreferenced_objects(ss::abort_source* as) {
     // remove on the state that has been persisted to cloud.
     const auto& s = stm_->state();
 
-    chunked_vector<object_id> to_remove;
+    chunked_vector<object_location> to_remove;
     for (const auto& [oid, obj_entry] : s.objects) {
         if (obj_entry.is_preregistration) {
             continue;
@@ -42,7 +42,13 @@ garbage_collector::remove_unreferenced_objects(ss::abort_source* as) {
 
         // TODO: split these into multiple updates in case we've got a lot of
         // objects to remove.
-        to_remove.emplace_back(oid);
+        to_remove.push_back(
+          object_location{
+            .id = oid,
+            .ts_path = obj_entry.imported_ts_location.transform(
+              [](const imported_ts_object_location& loc) {
+                  return loc.ts_path;
+              })});
         vlog(cd_log.debug, "Deleting L1 object: {}", oid);
     }
     if (to_remove.empty()) {
@@ -52,7 +58,12 @@ garbage_collector::remove_unreferenced_objects(ss::abort_source* as) {
     if (!del_res.has_value()) {
         co_return std::unexpected(error{"io error"});
     }
-    auto update_res = remove_objects_update::build(s, std::move(to_remove));
+    chunked_vector<object_id> remove_ids;
+    remove_ids.reserve(to_remove.size());
+    for (const auto& ext : to_remove) {
+        remove_ids.emplace_back(ext.id);
+    }
+    auto update_res = remove_objects_update::build(s, std::move(remove_ids));
     if (!update_res.has_value()) {
         co_return std::unexpected(error{"logic error"});
     }

src/v/cloud_topics/level_one/metastore/lsm/garbage_collector.cc

@@ -117,7 +117,7 @@ db_garbage_collector::remove_unreferenced_batch(
           std::move(object_range_res.error()), "Error getting object range"));
     }
     auto object_gen = object_range_res.value().get_rows();
-    chunked_vector<object_id> to_remove;
+    chunked_vector<object_location> to_remove;
     size_t batch_expire_count = 0;
     std::optional<object_id> next_batch_start{std::nullopt};
     while (auto obj_ref_opt = co_await object_gen()) {
@@ -178,7 +178,13 @@ db_garbage_collector::remove_unreferenced_batch(
           obj_entry.removed_data_size == obj_entry.total_data_size
           && obj_entry.last_updated <= deletion_delay_cutoff) {
             vlog(cd_log.debug, "Deleting L1 object: {}", oid);
-            to_remove.emplace_back(oid);
+            to_remove.push_back(
+              object_location{
+                .id = oid,
+                .ts_path = obj_entry.imported_ts_location.transform(
+                  [](const imported_ts_object_location& loc) {
+                      return loc.ts_path;
+                  })});
             if (to_remove.size() + batch_expire_count == batch_size) {
                 // Set an explicit next starting object if we had more than one
                 // batch of objects so we can make incremental progress.
@@ -206,7 +212,12 @@ db_garbage_collector::remove_unreferenced_batch(
           error(errc::io_error, "Error deleting objects: {}", del_res.error()));
     }
     probe_->gc_objects_deleted(num_to_remove);
-    remove_objects_db_update update{std::move(to_remove)};
+    chunked_vector<object_id> remove_ids;
+    remove_ids.reserve(to_remove.size());
+    for (const auto& ext : to_remove) {
+        remove_ids.emplace_back(ext.id);
+    }
+    remove_objects_db_update update{std::move(remove_ids)};
     chunked_vector<write_batch_row> rows;
     auto build_res = co_await update.build_rows(rows);
     if (!build_res.has_value()) {

src/v/cloud_topics/level_one/metastore/tests/garbage_collector_test.cc

@@ -129,6 +129,71 @@ class GarbageCollectorTest : public raft::stm_raft_fixture<simple_stm> {
         ASSERT_TRUE_CORO(repl_res.has_value());
     }
 
+    // Registers a single-extent imported object covering [base, last] whose
+    // backing segment is the TS-bucket object at ts_path (injected here).
+    // Unlike add_objects, an imported object has no L1 blob and is not
+    // preregistered.
+    ss::future<> add_imported_object(
+      simple_stm* stm,
+      model::topic_id_partition tp,
+      const ts_segment_path& ts_path,
+      kafka::offset base,
+      kafka::offset last) {
+        _io.put_ts_segment(ts_path, iobuf{});
+
+        new_object obj;
+        obj.oid = create_object_id();
+        obj.footer_pos = 0;
+        obj.object_size = 500;
+        obj.imported_ts_location = imported_ts_object_location{
+          .ts_path = ts_path};
+        obj.extent_metas[tp.topic_id][tp.partition] = new_object::metadata{
+          .base_offset = base,
+          .last_offset = last,
+          .max_timestamp = ts{10000},
+          .filepos = 0,
+          .len = 500,
+          .imported_ts_info = imported_ts_segment_info{},
+        };
+        chunked_vector<new_object> new_objects;
+        new_objects.push_back(std::move(obj));
+        term_state_update_t terms;
+        terms[tp].emplace_back(
+          term_start{.term_id = model::term_id{0}, .start_offset = base});
+
+        // Mark the partition migrating so the metastore adopts its log at the
+        // imported extents' (non-zero) base, then import via the normal
+        // add_objects path (imported objects skip pre-registration).
+        {
+            auto sync_res = co_await stm->sync(10s);
+            ASSERT_TRUE_CORO(sync_res.has_value());
+            set_migrating_update mig{.tp = tp, .migrating = true};
+            storage::record_batch_builder b(
+              model::record_batch_type::l1_stm, model::offset{0});
+            b.add_raw_kv(
+              serde::to_iobuf(set_migrating_update::key),
+              serde::to_iobuf(std::move(mig)));
+            auto r = co_await stm->replicate_and_wait(
+              sync_res.value(), std::move(b).build(), never_abort);
+            ASSERT_TRUE_CORO(r.has_value());
+        }
+
+        auto sync_res = co_await stm->sync(10s);
+        ASSERT_TRUE_CORO(sync_res.has_value());
+        add_objects_update update{
+          .new_objects = std::move(new_objects),
+          .new_terms = std::move(terms),
+        };
+        storage::record_batch_builder builder(
+          model::record_batch_type::l1_stm, model::offset{0});
+        builder.add_raw_kv(
+          serde::to_iobuf(add_objects_update::key),
+          serde::to_iobuf(std::move(update)));
+        auto repl_res = co_await stm->replicate_and_wait(
+          sync_res.value(), std::move(builder).build(), never_abort);
+        ASSERT_TRUE_CORO(repl_res.has_value());
+    }
+
     size_t count_objects() { return _io.list_objects().size(); }
 
     fake_io _io;
@@ -189,6 +254,30 @@ TEST_F(GarbageCollectorTest, TestGarbageCollectPartiallyRemovedObjects) {
     ASSERT_EQ(5, count_objects());
 }
 
+TEST_F(GarbageCollectorTest, TestGarbageCollectImportedObject) {
+    initialize_state_machines(1).get();
+    wait_for_leader(5s).get();
+    auto stm = get_stm<0>(*nodes().begin()->second);
+
+    auto tp = make_tp(0);
+    const ts_segment_path ts_path{"imported/seg-100-104.log"};
+    add_imported_object(stm.get(), tp, ts_path, o{100}, o{104}).get();
+    EXPECT_EQ(1, stm->state().objects.size());
+    EXPECT_TRUE(_io.has_ts_segment(ts_path));
+
+    // Trim past the imported extent so the object becomes unreferenced.
+    set_start_offset(stm.get(), tp, o{105}).get();
+
+    garbage_collector gc(stm.get(), &_io);
+    auto gc_res = gc.remove_unreferenced_objects(&never_abort).get();
+    ASSERT_TRUE(gc_res.has_value());
+
+    // The object row is gone, and its backing TS-bucket segment is deleted
+    // (routed by ts_path, not the object id).
+    EXPECT_EQ(0, stm->state().objects.size());
+    EXPECT_FALSE(_io.has_ts_segment(ts_path));
+}
+
 // A fake_io wrapper that fails on delete_objects
 class failing_io : public io {
 public:
@@ -217,8 +306,8 @@ class failing_io : public io {
           std::unexpected(errc::cloud_op_error));
     }
 
-    ss::future<std::expected<void, errc>>
-    delete_objects(chunked_vector<object_id>, ss::abort_source*) override {
+    ss::future<std::expected<void, errc>> delete_objects(
+      chunked_vector<object_location>, ss::abort_source*) override {
         return ss::make_ready_future<std::expected<void, errc>>(
           std::unexpected(errc::cloud_op_error));
     }