Merge pull request #29774 from redpanda-data/ct/core-15645/l0-gc-reset

[CORE-15645] Cloud Topics: Level Zero GC force reset (escape hatch)

Author: oleiman

proto/redpanda/core/admin/internal/cloud_topics/v1/level_zero.proto

@@ -25,22 +25,32 @@ option (pbgen.cpp_namespace) = "proto::admin::level_zero";
 // Admin endpoints to query Level-0 garbage collection workers for Cloud Topics.
 
 service LevelZeroService {
-    // Returns GC worker status for each shard on targeted node(s).
+    // Returns GC worker status for each shard on specified node (if present) or
+    // whole cluster.
     rpc GetStatus(GetStatusRequest) returns (GetStatusResponse) {
         option (pbgen.rpc) = {
             authz: SUPERUSER
         };
     }
 
-    // Starts GC workers on targeted node(s).
-    rpc Start(StartRequest) returns (StartResponse) {
+    // Starts GC workers on a specific node.
+    rpc StartGc(StartGcRequest) returns (StartGcResponse) {
         option (pbgen.rpc) = {
             authz: SUPERUSER
         };
     }
 
-    // Pauses GC workers on targeted node(s).
-    rpc Pause(PauseRequest) returns (PauseResponse) {
+    // Pauses GC workers on a specific node.
+    rpc PauseGc(PauseGcRequest) returns (PauseGcResponse) {
+        option (pbgen.rpc) = {
+            authz: SUPERUSER
+        };
+    }
+
+    // Resets GC worker internal state on the specified node.
+    // Drains in-flight operations and clears iteration state.
+    // GC resumes automatically if it was running before the reset.
+    rpc ResetGc(ResetGcRequest) returns (ResetGcResponse) {
         option (pbgen.rpc) = {
             authz: SUPERUSER
         };
@@ -95,41 +105,32 @@ message ShardStatus {
     Status status = 2;
 }
 
-// Request to start GC workers.
-message StartRequest {
-    // Target a specific node, or omit to start on all nodes.
+// Request to start GC workers on a specific node.
+message StartGcRequest {
+    // Target node. If omitted, the receiving node handles the request.
     optional int32 node_id = 1;
 }
 
-// Aggregated start results from one or more nodes.
-message StartResponse {
-    repeated StartResult results = 1;
-}
+// Empty on success; RPC error on failure.
+message StartGcResponse {}
 
-// Request to pause GC workers.
-message PauseRequest {
-    // Target a specific node, or omit to pause on all nodes.
+// Request to pause GC workers on a specific node.
+message PauseGcRequest {
+    // Target node. If omitted, the receiving node handles the request.
     optional int32 node_id = 1;
 }
 
-// Aggregated pause results from one or more nodes.
-message PauseResponse {
-    repeated PauseResult results = 1;
-}
+// Empty on success; RPC error on failure.
+message PauseGcResponse {}
 
-// Outcome of starting GC on a single node.
-message StartResult {
-    int32 node_id = 1;
-    // Set if the node could not be reached or returned an error.
-    optional string error = 3;
+// Request to reset GC worker state on a specific node.
+message ResetGcRequest {
+    // Target node. If omitted, the receiving node handles the request.
+    optional int32 node_id = 1;
 }
 
-// Outcome of pausing GC on a single node.
-message PauseResult {
-    int32 node_id = 1;
-    // Set if the node could not be reached or returned an error.
-    optional string error = 3;
-}
+// Empty on success; RPC error on failure.
+message ResetGcResponse {}
 
 // Request to advance a partition to the current cluster epoch.
 message AdvanceEpochRequest {
@@ -172,6 +173,7 @@ enum Status {
     L0_GC_STATUS_RUNNING = 2;
     L0_GC_STATUS_STOPPING = 3;
     L0_GC_STATUS_STOPPED = 4;
+    L0_GC_STATUS_RESETTING = 5;
 }
 
 // Request the L0 size estimate for a single partition.

src/v/cloud_topics/level_zero/gc/level_zero_gc.cc

@@ -28,9 +28,19 @@
 #include <seastar/core/sleep.hh>
 #include <seastar/coroutine/as_future.hh>
 
+#include <memory>
+
+namespace {
+constexpr ss::lowres_clock::duration control_timeout = 5s;
+}
+
 namespace cloud_topics {
 
 class level_zero_gc::list_delete_worker {
+    static constexpr auto handle_worker_exc = [](std::exception_ptr eptr) {
+        vlog(cd_log.warn, "Exception from delete worker: {}", eptr);
+    };
+
 public:
     explicit list_delete_worker(
       std::unique_ptr<object_storage> storage,
@@ -39,9 +49,7 @@ class level_zero_gc::list_delete_worker {
       : storage_(std::move(storage))
       , node_info_(std::move(node_info))
       , probe_(&probe)
-      , worker_([](std::exception_ptr eptr) {
-          vlog(cd_log.warn, "Exception from delete worker: {}", eptr);
-      }) {}
+      , worker_(std::make_unique<ssx::work_queue>(handle_worker_exc)) {}
     void start() {
         vlog(cd_log.info, "Starting cloud topics list/delete worker");
         if (as_.abort_requested()) {
@@ -63,11 +71,40 @@ class level_zero_gc::list_delete_worker {
         as_.request_abort();
         delete_sem_.broken();
         page_sem_.broken();
-        co_await worker_.shutdown();
+        co_await worker_->shutdown();
         co_await gate_.close();
         vlog(cd_log.info, "Stopped cloud topics list/delete worker");
     }
 
+    seastar::future<> reset() {
+        if (gate_.is_closed()) {
+            co_return;
+        }
+        vlog(cd_log.info, "Resetting cloud topics list/delete worker");
+
+        // Abort in-flight list/delete operations
+        as_.request_abort();
+
+        // Drain pending delete tasks
+        co_await worker_->shutdown();
+
+        // Wait for spawned delete fibers to complete
+        if (!gate_.is_closed()) {
+            co_await gate_.close();
+        }
+
+        continuation_token_.reset();
+        curr_prefix_.reset();
+        key_prefixes_.set_range(std::nullopt);
+
+        as_ = {};
+        gate_ = {};
+
+        worker_ = std::make_unique<ssx::work_queue>(handle_worker_exc);
+
+        vlog(cd_log.info, "Reset cloud topics list/delete worker");
+    }
+
     bool has_capacity() const { return page_sem_.available_units() > 0; }
 
     seastar::future<std::expected<
@@ -110,9 +147,9 @@ class level_zero_gc::list_delete_worker {
                 // unbounded.
                 u.emplace(seastar::consume_units(page_sem_, keys_total_bytes));
             }
-            worker_.submit([this,
-                            o = std::move(objects),
-                            u = std::move(u).value()]() mutable {
+            worker_->submit([this,
+                             o = std::move(objects),
+                             u = std::move(u).value()]() mutable {
                 return do_delete_objects(std::move(o), std::move(u));
             });
         }
@@ -222,7 +259,7 @@ class level_zero_gc::list_delete_worker {
     std::unique_ptr<object_storage> storage_;
     std::unique_ptr<node_info> node_info_;
     level_zero_gc_probe* probe_;
-    ssx::work_queue worker_;
+    std::unique_ptr<ssx::work_queue> worker_;
     // TODO: configurable limits?
     // max number of in-flight delete ops
     ssx::semaphore delete_sem_{5, "ct/gc/delete"};
@@ -621,14 +658,22 @@ level_zero_gc::level_zero_gc(
 
 level_zero_gc::~level_zero_gc() = default;
 
-void level_zero_gc::start() {
+seastar::future<> level_zero_gc::start() {
+    while (resetting_) {
+        co_await reset_cv_.wait(
+          control_timeout, [this] { return !resetting_; });
+    }
     vlog(cd_log.info, "Starting cloud topics L0 GC worker");
     delete_worker_->start();
     should_run_ = true;
     worker_cv_.signal();
 }
 
-void level_zero_gc::pause() {
+seastar::future<> level_zero_gc::pause() {
+    while (resetting_) {
+        co_await reset_cv_.wait(
+          control_timeout, [this] { return !resetting_; });
+    }
     vlog(cd_log.info, "Pausing cloud topics L0 GC worker");
     should_run_ = false;
     asrc_.request_abort();
@@ -645,13 +690,44 @@ seastar::future<> level_zero_gc::stop() {
     vlog(cd_log.info, "Stopped cloud_topics L0 GC worker");
 }
 
+seastar::future<> level_zero_gc::reset() {
+    if (should_shutdown_ || resetting_) {
+        co_return;
+    }
+    vlog(cd_log.info, "Resetting cloud topics L0 GC worker state");
+
+    resetting_ = true;
+    const bool was_running = should_run_;
+
+    auto done = ss::defer([this] {
+        resetting_ = false;
+        reset_cv_.broadcast();
+    });
+
+    // Pause the outer worker loop so it blocks on the CV
+    should_run_ = false;
+    asrc_.request_abort();
+
+    co_await delete_worker_->reset();
+
+    // Resume if was running, then clear the flag so that start()/pause()
+    // waiting on reset_cv_ don't race with the resume.
+    if (was_running && !should_shutdown_) {
+        delete_worker_->start();
+        should_run_ = true;
+        worker_cv_.signal();
+    }
+}
+
 std::string_view to_string_view(level_zero_gc::state s) {
     switch (s) {
         using enum level_zero_gc::state;
     case paused:
         return "level_zero_gc::state::paused";
     case running:
         return "level_zero_gc::state::running";
+    case resetting:
+        return "level_zero_gc::state::resetting";
     case stopping:
         return "level_zero_gc::state::stopping";
     case stopped:
@@ -667,6 +743,9 @@ auto level_zero_gc::get_state() const -> state {
         if (should_shutdown_) {
             return worker_.available() ? state::stopped : state::stopping;
         }
+        if (resetting_) {
+            return state::resetting;
+        }
         return should_run_ ? state::running : state::paused;
     }();
     vlog(cd_log.debug, "cloud_topics L0 GC worker state: {}", st);

src/v/cloud_topics/level_zero/gc/level_zero_gc.h

@@ -281,9 +281,18 @@ class level_zero_gc {
     /*
      * Request that GC be started or paused. These can be called multiple times
      * and in any order. The last invocation will eventually take effect.
+     *
+     * If a reset is in progress, these will block until it completes or
+     * control_timeout expires.
      */
-    void start();
-    void pause();
+    seastar::future<> start();
+    seastar::future<> pause();
+
+    /// Reset internal GC state without a full stop/start cycle.
+    /// Drains in-flight operations, clears pagination and prefix
+    /// iteration state, and prepares for a fresh collection sweep.
+    /// GC resumes automatically if it was running before the reset.
+    seastar::future<> reset();
 
     /*
      * Request and wait for GC to be completely stopped. After calling shutdown,
@@ -296,12 +305,14 @@ class level_zero_gc {
      *
      *   - paused: Paused indefinitely, call start() to run
      *   - running: GC will run until paused or stopped
+     *   - resetting: reset() is draining in-flight work
      *   - stopping: stop() requested but there may be work still in flight
      *   - stopped: Permanently stopped.
      */
     enum class state : uint8_t {
         paused,
         running,
+        resetting,
         stopping,
         stopped,
     };
@@ -317,9 +328,11 @@ class level_zero_gc {
 
     bool should_run_;
     bool should_shutdown_;
+    bool resetting_{false};
     seastar::abort_source asrc_;
     seastar::condition_variable worker_cv_;
     seastar::future<> worker_;
+    seastar::condition_variable reset_cv_;
 
     seastar::future<> worker();
     enum class collection_error : int8_t;

src/v/cloud_topics/level_zero/gc/tests/level_zero_gc_mt_test.cc

@@ -210,12 +210,13 @@ struct level_zero_gc_mt_test : public seastar_test {
     }
 
     // Add objects with various prefixes (call from shard 0 context)
-    void populate_objects(size_t count) {
+    void populate_objects(size_t count, bool dynamic_epoch = false) {
         g_bucket_state->objects.reserve(count);
         for (size_t i = 0; i < count; ++i) {
             auto prefix = static_cast<object_id::prefix_t>(i % 1000);
             auto id = object_id{
-              .epoch = cluster_epoch(1),
+              .epoch = cluster_epoch(
+                dynamic_epoch ? static_cast<int64_t>(i) : 1),
               .name = uuid_t::create(),
               .prefix = prefix,
             };
@@ -307,6 +308,38 @@ TEST_F_CORO(level_zero_gc_mt_test, no_eligible_epoch) {
     EXPECT_EQ(get_total_deleted(), 0);
 }
 
+/*
+ * Concurrent reset/start/pause cycles don't crash or corrupt state.
+ */
+TEST_F_CORO(level_zero_gc_mt_test, concurrent_reset_start_pause) {
+    populate_objects(num_objects, true /* dynamic_epoch */);
+    set_max_epoch(num_objects / 2 - 1);
+
+    co_await gc_.invoke_on_all(&level_zero_gc::start);
+    co_await ss::sleep(100ms);
+
+    std::vector<ss::future<>> futs;
+
+    for (int i = 0; i < 10; ++i) {
+        futs.push_back(
+          gc_.invoke_on_all([](level_zero_gc& gc) { return gc.reset(); }));
+        futs.push_back(
+          gc_.invoke_on_all([](level_zero_gc& gc) { return gc.reset(); }));
+        futs.push_back(gc_.invoke_on_all(&level_zero_gc::start));
+        futs.push_back(gc_.invoke_on_all(&level_zero_gc::pause));
+        futs.push_back(gc_.invoke_on_all(&level_zero_gc::start));
+    }
+
+    co_await ss::when_all_succeed(std::move(futs));
+
+    co_await gc_.invoke_on_all(&level_zero_gc::start);
+
+    set_max_epoch(num_objects);
+
+    RPTEST_REQUIRE_EVENTUALLY_CORO(
+      5s, [this] { return get_total_deleted() == num_objects; });
+}
+
 /*
  * Test start/pause/start cycle works correctly.
  */