ct/reconciler: integrate the metastore

src/v/cloud_topics/BUILD

@@ -94,11 +94,14 @@ redpanda_cc_library(
         ":data_plane_api",
         "//src/v/base",
         "//src/v/cloud_topics:state_accessors",
+        "//src/v/cloud_topics/level_one/common:file_io",
         "//src/v/cloud_topics/level_one/domain:domain_supervisor",
         "//src/v/cloud_topics/level_one/metastore:frontend",
+        "//src/v/cloud_topics/level_one/metastore:replicated_metastore",
         "//src/v/cloud_topics/level_zero/common:extent_meta",
         "//src/v/cloud_topics/reconciler",
         "//src/v/cluster",
+        "//src/v/config",
         "//src/v/container:chunked_vector",
         "//src/v/model",
         "//src/v/ssx:sharded_service_container",

src/v/cloud_topics/app.cc

@@ -13,8 +13,10 @@
 #include "cloud_topics/cluster_services.h"
 #include "cloud_topics/data_plane_api.h"
 #include "cloud_topics/data_plane_impl.h"
+#include "cloud_topics/level_one/metastore/replicated_metastore.h"
 #include "cluster/cluster_epoch_service.h"
 #include "cluster/controller.h"
+#include "config/node_config.h"
 #include "ssx/sharded_service_container.h"
 
 #include <seastar/core/coroutine.hh>
@@ -49,8 +51,17 @@ ss::future<> app::construct(
 
     co_await construct_service(state, data_plane.get());
 
+    // Ensure the L1 staging directory exists before creating the l1_io service
+    co_await ss::recursive_touch_directory(
+      config::node().l1_staging_path().string());
+
     co_await construct_service(
-      reconciler, partition_mgr, remote, data_plane.get(), bucket);
+      l1_io,
+      config::node().l1_staging_path(),
+      ss::sharded_parameter([&remote] { return &remote->local(); }),
+      bucket,
+      ss::sharded_parameter([&cloud_cache] { return &cloud_cache->local(); }));
+
     co_await construct_service(domain_supervisor, controller);
     co_await construct_service(
       l1_metastore_fe,
@@ -60,6 +71,18 @@ ss::future<> app::construct(
       shard_table,
       connection_cache,
       &domain_supervisor);
+
+    co_await construct_service(
+      reconciler,
+      partition_mgr,
+      data_plane.get(),
+      ss::sharded_parameter([this] { return &l1_io.local(); }),
+      ss::sharded_parameter([this] {
+          return std::make_unique<l1::replicated_metastore>(
+            l1_metastore_fe.local());
+      }),
+      ss::sharded_parameter(
+        [&metadata_cache] { return &metadata_cache->local(); }));
 }
 
 ss::future<> app::start() {

src/v/cloud_topics/app.h

@@ -10,6 +10,7 @@
 
 #pragma once
 
+#include "cloud_topics/level_one/common/file_io.h"
 #include "cloud_topics/level_one/domain/domain_supervisor.h"
 #include "cloud_topics/level_one/metastore/frontend.h"
 #include "cloud_topics/reconciler/reconciler.h"
@@ -71,6 +72,7 @@ class app : public ssx::sharded_service_container {
     ss::sstring _logger_name;
     std::unique_ptr<data_plane_api> data_plane;
     ss::sharded<state_accessors> state;
+    ss::sharded<l1::file_io> l1_io;
     ss::sharded<reconciler::reconciler> reconciler;
     ss::sharded<l1::domain_supervisor> domain_supervisor;
     ss::sharded<l1::frontend> l1_metastore_fe;

src/v/cloud_topics/level_one/common/BUILD

@@ -4,13 +4,21 @@ package(default_visibility = ["//src/v/cloud_topics/level_one:__subpackages__"])
 
 redpanda_cc_library(
     name = "object_id",
+    srcs = [
+        "object_id.cc",
+    ],
     hdrs = [
         "object_id.h",
     ],
+    visibility = [
+        "//src/v/cloud_topics/level_one:__subpackages__",
+        "//src/v/cloud_topics/reconciler:__subpackages__",
+    ],
     deps = [
         "//src/v/base",
         "//src/v/utils:named_type",
         "//src/v/utils:uuid",
+        "@fmt",
     ],
 )
 
@@ -48,6 +56,10 @@ redpanda_cc_library(
         "//src/v/serde:vector",
         "//src/v/storage:record_batch_builder",
     ],
+    visibility = [
+        "//src/v/cloud_topics/level_one:__subpackages__",
+        "//src/v/cloud_topics/reconciler:__subpackages__",
+    ],
     deps = [
         "//src/v/base",
         "//src/v/container:chunked_vector",
@@ -63,6 +75,10 @@ redpanda_cc_library(
     name = "abstract_io",
     srcs = ["abstract_io.cc"],
     hdrs = ["abstract_io.h"],
+    visibility = [
+        "//src/v/cloud_topics/level_one:__subpackages__",
+        "//src/v/cloud_topics/reconciler:__subpackages__",
+    ],
     deps = [
         ":object_id",
         "//src/v/container:chunked_vector",
@@ -75,6 +91,11 @@ redpanda_cc_library(
     name = "file_io",
     srcs = ["file_io.cc"],
     hdrs = ["file_io.h"],
+    visibility = [
+        "//src/v/cloud_topics:__pkg__",
+        "//src/v/cloud_topics/level_one:__subpackages__",
+        "//src/v/cloud_topics/reconciler:__pkg__",
+    ],
     deps = [
         ":abstract_io",
         ":object_id",

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

@@ -49,6 +49,8 @@ namespace cloud_topics::l1 {
 // side effects. As such, callers can think of this interface as thread safe.
 class metastore {
 public:
+    virtual ~metastore() = default;
+
     enum class errc {
         missing_ntp,
         invalid_request,

src/v/cloud_topics/reconciler/BUILD

@@ -3,16 +3,16 @@ load("//bazel:build.bzl", "redpanda_cc_library")
 package(default_visibility = [":__subpackages__"])
 
 redpanda_cc_library(
-    name = "range_batch_consumer",
+    name = "reconciliation_consumer",
     srcs = [
-        "range_batch_consumer.cc",
+        "reconciliation_consumer.cc",
     ],
     hdrs = [
-        "range_batch_consumer.h",
+        "reconciliation_consumer.h",
     ],
     deps = [
         "//src/v/base",
-        "//src/v/bytes:iobuf",
+        "//src/v/cloud_topics/level_one/common:object",
         "//src/v/model",
         "@abseil-cpp//absl/container:btree",
         "@seastar",
@@ -33,14 +33,17 @@ redpanda_cc_library(
     ],
     visibility = ["//visibility:public"],
     deps = [
-        ":range_batch_consumer",
+        ":reconciliation_consumer",
         "//src/v/base",
-        "//src/v/cloud_io:remote",
-        "//src/v/cloud_storage",
         "//src/v/cloud_storage_clients",
         "//src/v/cloud_topics:object_utils",
         "//src/v/cloud_topics:types",
+        "//src/v/cloud_topics/level_one/common:file_io",
+        "//src/v/cloud_topics/level_one/common:object",
+        "//src/v/cloud_topics/level_one/common:object_id",
         "//src/v/cloud_topics/level_one/common:object_utils",
+        "//src/v/cloud_topics/level_one/metastore",
+        "//src/v/cloud_topics/level_one/metastore:replicated_metastore",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm_api",
         "//src/v/cluster",
         "//src/v/cluster:notification",
@@ -49,6 +52,7 @@ redpanda_cc_library(
         "//src/v/model",
         "//src/v/random:generators",
         "@abseil-cpp//absl/container:btree",
+        "@abseil-cpp//absl/container:flat_hash_map",
         "@abseil-cpp//absl/container:node_hash_map",
         "@seastar",
     ],