ct: reorganize cloud_topics::app

This reorganizes cloud_topics::app so that:
1. it is a single non-sharded entity,
2. it includes all cloud_topics-related sharded services

Previously, when cloud_topics::app was sharded, #2 was difficult because
there are some cloud_topics-related services that need to be sharded,
but app itself was sharded in application.cc; having a sharded service
own another sharded service seems like a bad idea.

In doing this, I needed to wrap the data_plane_api in another class in
order to shard it, which I've done with the recently added
state_accessors. This lets us continue to use the data_plane_api instead
of the data_plane_impl, as we were before, as a sharded service.

Using the state_accessors also lets us plumb the data plane API through
to cluster::partition and to the partition_proxy without introducing a
circular dependency. Previously we plumbed down cloud_topics::app, which
made it easy to trip over dependencies.

Author: andrwng

src/v/cloud_topics/BUILD

@@ -77,6 +77,10 @@ redpanda_cc_library(
     hdrs = [
         "app.h",
     ],
+    implementation_deps = [
+        ":cluster_services_interface",
+        ":data_plane_impl",
+    ],
     include_prefix = "cloud_topics",
     visibility = [
         "//src/v/cloud_topics/frontend:__pkg__",
@@ -87,11 +91,14 @@ redpanda_cc_library(
     ],
     deps = [
         "//src/v/base",
-        "//src/v/cloud_topics:data_plane_api",
+        "//src/v/cloud_topics:state_accessors",
         "//src/v/cloud_topics/level_one/domain:domain_supervisor",
+        "//src/v/cloud_topics/level_one/metastore:frontend",
         "//src/v/cloud_topics/level_zero/common:extent_meta",
+        "//src/v/cluster",
         "//src/v/container:chunked_vector",
         "//src/v/model",
+        "//src/v/ssx:sharded_service_container",
         "//src/v/storage",
         "@seastar",
     ],

src/v/cloud_topics/app.cc

@@ -10,30 +10,91 @@
 
 #include "cloud_topics/app.h"
 
+#include "cloud_topics/cluster_services.h"
+#include "cloud_topics/data_plane_impl.h"
+#include "cluster/cluster_epoch_service.h"
+#include "cluster/controller.h"
+#include "ssx/sharded_service_container.h"
+
 #include <seastar/core/coroutine.hh>
 
 namespace experimental::cloud_topics {
 
-app::app(
-  ss::shared_ptr<data_plane_api> dp,
-  std::unique_ptr<l1::domain_supervisor> l1_cp)
-  : _data_plane(std::move(dp))
-  , _domain_supervisor(std::move(l1_cp)) {}
+namespace {
+class real_cluster_services
+  : public experimental::cloud_topics::cluster_services {
+public:
+    explicit real_cluster_services(
+      ss::sharded<cluster::cluster_epoch_service<>>* epoch_generator)
+      : _epoch_service(epoch_generator) {}
+
+    seastar::future<experimental::cloud_topics::cluster_epoch>
+    current_epoch(seastar::abort_source* as) override {
+        std::expected<int64_t, std::error_code> epoch
+          = co_await _epoch_service->local().get_cached_epoch(as);
+        if (!epoch) {
+            throw std::system_error(epoch.error());
+        }
+        co_return experimental::cloud_topics::cluster_epoch(epoch.value());
+    }
+
+private:
+    ss::sharded<cluster::cluster_epoch_service<>>* _epoch_service;
+};
+} // namespace
 
-seastar::future<> app::start() {
-    co_await _domain_supervisor->start();
-    co_await _data_plane->start();
+app::app()
+  : ssx::sharded_service_container("cloud_topics::app") {}
+
+ss::future<> app::construct(
+  model::node_id self,
+  cluster::controller* controller,
+  ss::sharded<cluster::partition_manager>* partition_mgr,
+  ss::sharded<cluster::partition_leaders_table>* leaders_table,
+  ss::sharded<cluster::shard_table>* shard_table,
+  ss::sharded<cloud_io::remote>* remote,
+  ss::sharded<cloud_storage::cache>* cloud_cache,
+  ss::sharded<cluster::metadata_cache>* metadata_cache,
+  ss::sharded<rpc::connection_cache>* connection_cache,
+  cloud_storage_clients::bucket_name bucket,
+  ss::sharded<storage::api>* storage) {
+    co_await construct_service(
+      state,
+      ss::sharded_parameter([remote, cloud_cache, bucket, storage, controller] {
+          return experimental::cloud_topics::make_data_plane(
+            remote,
+            cloud_cache,
+            bucket,
+            storage,
+            std::make_unique<real_cluster_services>(
+              &controller->get_cluster_epoch_generator()));
+      }));
+    co_await construct_service(domain_supervisor, controller);
+    co_await construct_service(
+      l1_metastore_fe,
+      self,
+      metadata_cache,
+      leaders_table,
+      shard_table,
+      connection_cache,
+      &domain_supervisor);
 }
 
-seastar::future<> app::stop() {
-    co_await _domain_supervisor->stop();
-    co_await _data_plane->stop();
+ss::future<> app::start() {
+    co_await state.invoke_on_all([](auto& s) { return s.start(); });
+    co_await domain_supervisor.invoke_on_all(
+      [](auto& ds) { return ds.start(); });
 }
 
-ss::shared_ptr<data_plane_api> app::get_data_plane_api() { return _data_plane; }
+ss::future<> app::stop() {
+    ssx::sharded_service_container::shutdown();
+    co_return;
+}
 
-l1::domain_supervisor* app::get_l1_domain_supervisor() {
-    return _domain_supervisor.get();
+ss::sharded<l1::frontend>* app::get_sharded_l1_metastore_fe() {
+    return &l1_metastore_fe;
 }
 
+ss::sharded<state_accessors>* app::get_state() { return &state; }
+
 } // namespace experimental::cloud_topics

src/v/cloud_topics/app.h

@@ -10,39 +10,64 @@
 
 #pragma once
 
-#include "cloud_topics/data_plane_api.h"
 #include "cloud_topics/level_one/domain/domain_supervisor.h"
+#include "cloud_topics/level_one/metastore/frontend.h"
+#include "cloud_topics/state_accessors.h"
+#include "ssx/sharded_service_container.h"
 
 #include <seastar/core/future.hh>
 #include <seastar/core/lowres_clock.hh>
 #include <seastar/core/sharded.hh>
 
+namespace cloud_io {
+class remote;
+} // namespace cloud_io
+namespace cloud_storage {
+class cache;
+} // namespace cloud_storage
+namespace storage {
+class api;
+} // namespace storage
+
 namespace experimental::cloud_topics {
 
-// Simple container to use with seastar::sharded.
-// The seastar::sharded wants to know the size of the object at compile time.
-class app {
+class app : public ssx::sharded_service_container {
 public:
-    explicit app(
-      ss::shared_ptr<data_plane_api>, std::unique_ptr<l1::domain_supervisor>);
+    explicit app();
 
     app(const app&) = delete;
     app& operator=(const app&) = delete;
     app(app&&) noexcept = delete;
     app& operator=(app&&) noexcept = delete;
     ~app() = default;
 
-    seastar::future<> start();
-    seastar::future<> stop();
+    ss::future<> construct(
+      model::node_id,
+      cluster::controller*,
+      ss::sharded<cluster::partition_manager>*,
+      ss::sharded<cluster::partition_leaders_table>*,
+      ss::sharded<cluster::shard_table>*,
+      ss::sharded<cloud_io::remote>*,
+      ss::sharded<cloud_storage::cache>*,
+      ss::sharded<cluster::metadata_cache>*,
+      ss::sharded<rpc::connection_cache>*,
+      cloud_storage_clients::bucket_name,
+      ss::sharded<storage::api>*);
+
+    ss::future<> start();
+
+    // Call stop on each sharded service and call their destructors.
+    ss::future<> stop();
 
-    ss::shared_ptr<data_plane_api> get_data_plane_api();
-    l1::domain_supervisor* get_l1_domain_supervisor();
+    ss::sharded<l1::frontend>* get_sharded_l1_metastore_fe();
+    ss::sharded<state_accessors>* get_state();
 
     // TODO: add 'get_control_plane_api' etc
 
 private:
-    ss::shared_ptr<data_plane_api> _data_plane;
-    std::unique_ptr<l1::domain_supervisor> _domain_supervisor;
+    ss::sharded<state_accessors> state;
+    ss::sharded<l1::domain_supervisor> domain_supervisor;
+    ss::sharded<l1::frontend> l1_metastore_fe;
 };
 
 } // namespace experimental::cloud_topics

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

@@ -222,13 +222,13 @@ frontend::frontend(
   ss::sharded<cluster::partition_leaders_table>* leaders,
   ss::sharded<cluster::shard_table>* shards,
   ss::sharded<::rpc::connection_cache>* connections,
-  domain_supervisor* domain_supervisor)
+  ss::sharded<domain_supervisor>* domain_supervisor)
   : _self(self)
   , _metadata(metadata)
   , _leaders(leaders)
   , _shard_table(shards)
   , _connection_cache(connections)
-  , _domain_supervisor(domain_supervisor) {}
+  , _domain_supervisor(&domain_supervisor->local()) {}
 
 ss::future<> frontend::stop() { return _gate.close(); }
 

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

@@ -51,7 +51,7 @@ class frontend : public ss::peering_sharded_service<frontend> {
       ss::sharded<cluster::partition_leaders_table>*,
       ss::sharded<cluster::shard_table>*,
       ss::sharded<::rpc::connection_cache>*,
-      domain_supervisor*);
+      ss::sharded<domain_supervisor>*);
 
     ss::future<> stop();
 

src/v/cluster/BUILD

@@ -683,6 +683,7 @@ redpanda_cc_library(
         "//src/v/cloud_storage:topic_mount_manifest_path",
         "//src/v/cloud_storage:types",
         "//src/v/cloud_storage_clients",
+        "//src/v/cloud_topics:state_accessors",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm_api",
         "//src/v/cluster_link/model",

src/v/cluster/partition.cc

@@ -53,9 +53,9 @@ partition::partition(
   ss::sharded<features::feature_table>& feature_table,
   ss::sharded<archival::upload_housekeeping_service>& upload_hks,
   std::optional<cloud_storage_clients::bucket_name> read_replica_bucket,
-  ss::sharded<experimental::cloud_topics::app>& ct_app)
+  ss::sharded<experimental::cloud_topics::state_accessors>* ct_state)
   : _raft(std::move(r))
-  , _cloud_topics_app(ct_app)
+  , _cloud_topics_state(ct_state)
   , _probe(std::make_unique<replicated_partition_probe>(*this))
   , _feature_table(feature_table)
   , _archival_conf(std::move(archival_conf))
@@ -1856,9 +1856,9 @@ ss::future<result<ssx::rwlock_unit>> partition::hold_writes_enabled() {
     co_return *std::move(maybe_units);
 }
 
-ss::sharded<experimental::cloud_topics::app>&
-partition::get_cloud_topics_data_api() noexcept {
-    return _cloud_topics_app;
+ss::sharded<experimental::cloud_topics::state_accessors>*
+partition::get_cloud_topics_state() noexcept {
+    return _cloud_topics_state;
 }
 
 } // namespace cluster

src/v/cluster/partition.h

@@ -32,7 +32,7 @@
 
 namespace experimental::cloud_topics {
 class ctp_stm_api;
-class app;
+class state_accessors;
 }; // namespace experimental::cloud_topics
 
 namespace cluster {
@@ -58,7 +58,7 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
       ss::sharded<features::feature_table>&,
       ss::sharded<archival::upload_housekeeping_service>&,
       std::optional<cloud_storage_clients::bucket_name> read_replica_bucket,
-      ss::sharded<experimental::cloud_topics::app>& ct_app);
+      ss::sharded<experimental::cloud_topics::state_accessors>* ct_state);
 
     ~partition() = default;
 
@@ -403,8 +403,8 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
 
     // Returns a pointer to the data plane api if cloud topics is enabled for
     // this partition. Otherwise, nullopt is returned
-    ss::sharded<experimental::cloud_topics::app>&
-    get_cloud_topics_data_api() noexcept;
+    ss::sharded<experimental::cloud_topics::state_accessors>*
+    get_cloud_topics_state() noexcept;
 
 private:
     ss::future<>
@@ -432,7 +432,8 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
     ss::shared_ptr<archival_metadata_stm> _archival_meta_stm;
     ss::shared_ptr<partition_properties_stm> _partition_properties_stm;
     ss::shared_ptr<experimental::cloud_topics::ctp_stm_api> _ctp_stm_api;
-    ss::sharded<experimental::cloud_topics::app>& _cloud_topics_app;
+    ss::sharded<experimental::cloud_topics::state_accessors>*
+      _cloud_topics_state;
     ss::abort_source _as;
     partition_probe _probe;
     ss::sharded<features::feature_table>& _feature_table;

src/v/cluster/partition_manager.cc

@@ -60,7 +60,7 @@ partition_manager::partition_manager(
   ss::sharded<features::feature_table>& feature_table,
   ss::sharded<archival::upload_housekeeping_service>& upload_hks,
   config::binding<std::chrono::milliseconds> partition_shutdown_timeout,
-  ss::sharded<experimental::cloud_topics::app>& cloud_topics_app)
+  ss::sharded<experimental::cloud_topics::state_accessors>* cloud_topics_state)
   : _storage(storage.local())
   , _raft_manager(raft)
   , _partition_recovery_mgr(recovery_mgr)
@@ -70,7 +70,7 @@ partition_manager::partition_manager(
   , _feature_table(feature_table)
   , _upload_hks(upload_hks)
   , _partition_shutdown_timeout(std::move(partition_shutdown_timeout))
-  , _cloud_topics_app(cloud_topics_app) {
+  , _cloud_topics_state(cloud_topics_state) {
     _leader_notify_handle
       = _raft_manager.local().register_leadership_notification(
         [this](
@@ -302,7 +302,7 @@ ss::future<consensus_ptr> partition_manager::manage(
       _feature_table,
       _upload_hks,
       read_replica_bucket,
-      _cloud_topics_app);
+      _cloud_topics_state);
 
     _ntp_table.emplace(log->config().ntp(), p);
     _raft_table.emplace(group, p);

src/v/cluster/partition_manager.h

@@ -33,7 +33,7 @@
 #include <chrono>
 
 namespace experimental::cloud_topics {
-class app;
+class state_accessors;
 }
 
 namespace cluster {
@@ -53,7 +53,7 @@ class partition_manager
       ss::sharded<features::feature_table>&,
       ss::sharded<archival::upload_housekeeping_service>&,
       config::binding<std::chrono::milliseconds>,
-      ss::sharded<experimental::cloud_topics::app>&);
+      ss::sharded<experimental::cloud_topics::state_accessors>*);
 
     ~partition_manager();
 
@@ -308,7 +308,8 @@ class partition_manager
     state_machine_registry _stm_registry;
 
     // The sharded app may not be initialized if cloud topics isn't enabled.
-    ss::sharded<experimental::cloud_topics::app>& _cloud_topics_app;
+    ss::sharded<experimental::cloud_topics::state_accessors>*
+      _cloud_topics_state;
 
     friend std::ostream& operator<<(std::ostream&, const partition_manager&);
     friend std::ostream& operator<<(