ct/l1: fix unmanage_cb_t and some life-time issues

This `unmanage_cb_t` type is typedef'd as returning `void`, when actually
it was invoking a function returning an `ss::future<>`, due to some asynchronous
work that must occur when unmanaging a partition.

Fix the `compaction_scheduler::unmanage_partition()` implementation and some
of the functions in the call-stack so that the asynchronous work is pushed
to a detached fiber lower in `worker_manager`. To ensure proper lifetimes,
add a `ss::gate` to the `worker_manager` and use this to prevent any requests
from entering the `worker_manager` while it is being/after it is stopped.

Author: WillemKauf

src/v/cloud_topics/level_one/compaction/log_collector.cc

@@ -66,7 +66,7 @@ ss::future<> partition_leader_log_collector::stop_collecting_logs() {
 }
 
 void partition_leader_log_collector::on_ntp_change(
-  cluster::topic_table::ntp_delta delta) {
+  const cluster::topic_table::ntp_delta& delta) {
     auto& ntp = delta.ntp;
     auto is_managed = _is_managed_cb(ntp);
 
@@ -75,7 +75,7 @@ void partition_leader_log_collector::on_ntp_change(
     case delta_type::removed: {
         // Partition/possibly topic was removed. Unmanage it if necessary.
         if (is_managed) {
-            _unmanage_cb(std::move(ntp), "Partition removed");
+            _unmanage_cb(ntp, "Partition removed");
         }
         return;
     }
@@ -104,7 +104,7 @@ void partition_leader_log_collector::on_ntp_change(
         if (!is_compacted_cloud_topic && is_managed) {
             // This is likely an existing cloud topic which is no longer
             // `compact` enabled.
-            _unmanage_cb(std::move(ntp), "Disabled compaction");
+            _unmanage_cb(ntp, "Disabled compaction");
         }
         return;
     }

src/v/cloud_topics/level_one/compaction/log_collector.h

@@ -102,7 +102,7 @@ class partition_leader_log_collector : public log_collector {
     // Register operations can be performed synchronously while unregister
     // operations are performed in a backgrounded fiber (see
     // `compaction_scheduler::unmanage_partition()`).
-    void on_ntp_change(cluster::topic_table::ntp_delta);
+    void on_ntp_change(const cluster::topic_table::ntp_delta&);
 
     // Registers/unregisters `ntp`s with the `compaction_scheduler` using
     // leadership notifications from the `partition_leaders_table`. The

src/v/cloud_topics/level_one/compaction/scheduler.cc

@@ -34,8 +34,8 @@ compaction_scheduler::compaction_scheduler(
         const model::ntp& ntp,
         const model::topic_id_partition& tidp,
         std::string_view ctx) { manage_partition(ntp, tidp, ctx); },
-      [this](model::ntp ntp, std::string_view ctx) {
-          return unmanage_partition(std::move(ntp), ctx);
+      [this](const model::ntp& ntp, std::string_view ctx) {
+          unmanage_partition(ntp, ctx);
       },
       [this](const model::ntp& ntp) { return is_managed(ntp); },
       state))
@@ -96,8 +96,8 @@ void compaction_scheduler::manage_partition(
     _probe.set_log_count(_logs.size());
 }
 
-ss::future<>
-compaction_scheduler::unmanage_partition(model::ntp ntp, std::string_view ctx) {
+void compaction_scheduler::unmanage_partition(
+  const model::ntp& ntp, std::string_view ctx) {
     auto tidp_entry = _ntp_to_tidp.extract(ntp);
     if (!tidp_entry.has_value()) {
         vassert(
@@ -132,7 +132,7 @@ compaction_scheduler::unmanage_partition(model::ntp ntp, std::string_view ctx) {
     // Request that compaction of this CTP be stopped, if in flight. `handle` is
     // a `lw_shared_ptr`- we can allow it to go out of scope here without fear
     // of UAF elsewhere.
-    co_await _worker_manager.request_stop_compaction(handle);
+    _worker_manager.request_stop_compaction(std::move(handle));
     _probe.set_log_count(_logs.size());
 }
 

src/v/cloud_topics/level_one/compaction/scheduler.h

@@ -73,9 +73,9 @@ class compaction_scheduler {
 
     // Removes the `tidp` from the list of managed partitions. No-ops if the
     // provided `tidp` is not managed by this scheduler. Because the `tidp`
-    // may be undergoing an inflight compaction, this function will block until
-    // it is complete (an early stop is requested by this function).
-    ss::future<> unmanage_partition(model::ntp, std::string_view);
+    // may be undergoing an inflight compaction, an early stop is requested by
+    // the `_worker_manager`.
+    void unmanage_partition(const model::ntp&, std::string_view);
 
 private:
     // Starts the backgrounded scheduling loop.

src/v/cloud_topics/level_one/compaction/tests/scheduler_multithread_test.cc

@@ -54,16 +54,15 @@ TEST_F(SchedulerTestFixture, TestSchedulerMultithread) {
       };
     auto unmanage_random_partition_func = [this, &managed_ntps]() {
         if (managed_ntps.empty()) {
-            return ss::now();
+            return;
         }
         auto ntp_to_remove
           = random_generators::random_choice(
               std::vector<std::pair<model::ntp, model::topic_id_partition>>(
                 managed_ntps.begin(), managed_ntps.end()))
               .first;
         managed_ntps.erase(ntp_to_remove);
-        return scheduler->unmanage_partition(
-          std::move(ntp_to_remove), "unmanage_partition_func");
+        scheduler->unmanage_partition(ntp_to_remove, "unmanage_partition_func");
     };
     auto pause_random_worker_func = [this, &paused_workers]() {
         auto random_shard = random_generators::get_int(ss::smp::count - 1);
@@ -109,8 +108,8 @@ TEST_F(SchedulerTestFixture, TestSchedulerMultithread) {
     auto unmanage_ntp_fut = ss::do_until(
       [&] { return as.abort_requested(); },
       [&]() {
-          return unmanage_random_partition_func().then(
-            [&]() { return ss::sleep(unmanage_sleep); });
+          unmanage_random_partition_func();
+          return ss::sleep(unmanage_sleep);
       });
 
     static constexpr auto pause_worker_sleep = 50ms;

src/v/cloud_topics/level_one/compaction/tests/scheduler_test.cc

@@ -56,16 +56,15 @@ TEST_F(SchedulerTestFixture, TestScheduler) {
       };
     auto unmanage_random_partition_func = [this, &managed_ntps]() {
         if (managed_ntps.empty()) {
-            return ss::now();
+            return;
         }
         auto ntp_to_remove
           = random_generators::random_choice(
               std::vector<std::pair<model::ntp, model::topic_id_partition>>(
                 managed_ntps.begin(), managed_ntps.end()))
               .first;
         managed_ntps.erase(ntp_to_remove);
-        return scheduler->unmanage_partition(
-          std::move(ntp_to_remove), "unmanage_partition_func");
+        scheduler->unmanage_partition(ntp_to_remove, "unmanage_partition_func");
     };
     auto pause_random_worker_func = [this, &paused_workers]() {
         auto random_shard = random_generators::get_int(ss::smp::count - 1);
@@ -132,8 +131,8 @@ TEST_F(SchedulerTestFixture, TestScheduler) {
     auto unmanage_ntp_fut = ss::do_until(
       [&] { return as.abort_requested(); },
       [&]() {
-          return unmanage_random_partition_func().then(
-            [&]() { return ss::sleep(unmanage_sleep); });
+          unmanage_random_partition_func();
+          return ss::sleep(unmanage_sleep);
       });
 
     static constexpr auto pause_worker_sleep = 50ms;

src/v/cloud_topics/level_one/compaction/worker_manager.cc

@@ -16,6 +16,7 @@
 #include "cloud_topics/level_one/compaction/worker.h"
 #include "cloud_topics/level_one/metastore/replicated_metastore.h"
 #include "model/fundamental.h"
+#include "ssx/future-util.h"
 
 namespace cloud_topics::l1 {
 
@@ -43,7 +44,10 @@ ss::future<> worker_manager::start() {
     co_await _workers.invoke_on_all(&compaction_worker::start);
 }
 
-ss::future<> worker_manager::stop() { co_await _workers.stop(); }
+ss::future<> worker_manager::stop() {
+    co_await _gate.close();
+    co_await _workers.stop();
+}
 
 std::optional<foreign_log_compaction_meta_ptr>
 worker_manager::try_acquire_work(ss::shard_id shard) {
@@ -93,35 +97,39 @@ void worker_manager::complete_work(log_compaction_meta* log) {
     _probe.log_compacted();
 }
 
-ss::future<>
-worker_manager::request_stop_compaction(log_compaction_meta_ptr log) {
+void worker_manager::request_stop_compaction(log_compaction_meta_ptr log) {
     if (!log) {
-        co_return;
+        return;
     }
 
     auto shard_opt = log->inflight_shard;
     if (!shard_opt.has_value()) {
-        co_return;
+        return;
     }
 
     auto shard = shard_opt.value();
 
-    co_await _workers.invoke_on(shard, [](compaction_worker& worker) {
-        return worker.terminate_current_job();
+    ssx::spawn_with_gate(_gate, [this, shard]() {
+        return _workers.invoke_on(shard, [](compaction_worker& worker) {
+            return worker.terminate_current_job();
+        });
     });
 }
 
 ss::future<> worker_manager::alert_workers() {
+    auto guard = _gate.hold();
     co_await _workers.invoke_on_all(
       [](compaction_worker& worker) { worker.alert_worker(); });
 }
 
 ss::future<> worker_manager::pause_worker(ss::shard_id worker) {
+    auto guard = _gate.hold();
     co_await _workers.invoke_on(
       worker, [](compaction_worker& worker) { return worker.pause_worker(); });
 }
 
 ss::future<> worker_manager::resume_worker(ss::shard_id worker) {
+    auto guard = _gate.hold();
     co_await _workers.invoke_on(
       worker, [](compaction_worker& worker) { return worker.resume_worker(); });
 }

src/v/cloud_topics/level_one/compaction/worker_manager.h

@@ -77,7 +77,7 @@ class worker_manager {
     // compaction jobs to be ran. This function is ideally used when e.g. a
     // partition is removed or the `cleanup.policy` for a topic is changed and a
     // single compaction job must be stopped.
-    ss::future<> request_stop_compaction(log_compaction_meta_ptr);
+    void request_stop_compaction(log_compaction_meta_ptr);
 
     // Alert all workers that new jobs have become available in the
     // `_work_queue`.
@@ -112,6 +112,8 @@ class worker_manager {
 
     // A sharded pool of compaction workers.
     ss::sharded<compaction_worker> _workers;
+
+    ss::gate _gate;
 };
 
 } // namespace cloud_topics::l1