ct/l1: use l1_reader_cache in the fetch path

Wire the reader cache into frontend::make_l1_reader. On each L1 fetch
the frontend first checks the cache for a reader matching the
topic_id_partition and start offset. On a hit the positioned reader is
reused, skipping the metastore RPC, footer fetch, and stream open. On a
miss a fresh reader is constructed and handed to the cache so it will be
returned to the idle pool when the fetch completes.

Author: wdberkeley

src/v/cloud_topics/frontend/BUILD

@@ -26,6 +26,7 @@ redpanda_cc_library(
         "//src/v/cloud_storage:types",
         "//src/v/cloud_topics:logger",
         "//src/v/cloud_topics:topic_id_partition",
+        "//src/v/cloud_topics/level_one/frontend_reader:l1_reader_cache",
         "//src/v/cloud_topics/level_one/metastore",
         "//src/v/cloud_topics/level_zero/common:producer_queue",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm",

src/v/cloud_topics/frontend/frontend.cc

@@ -12,6 +12,7 @@
 #include "cloud_storage/types.h"
 #include "cloud_topics/data_plane_api.h"
 #include "cloud_topics/frontend/errc.h"
+#include "cloud_topics/level_one/frontend_reader/l1_reader_cache.h"
 #include "cloud_topics/level_one/frontend_reader/level_one_reader.h"
 #include "cloud_topics/level_one/metastore/metastore.h"
 #include "cloud_topics/level_zero/common/extent_meta.h"
@@ -287,8 +288,7 @@ frontend::make_reader(cloud_topic_log_reader_config cfg) {
         if (!tidp) {
             throw topic_config_not_found_exception(ntp());
         }
-        co_return storage::translating_reader{
-          model::record_batch_reader(make_l1_reader(cfg, *tidp))};
+        co_return storage::translating_reader{make_l1_reader(cfg, *tidp)};
     }
     co_return storage::translating_reader{
       model::record_batch_reader(make_l0_reader(cfg)),
@@ -387,16 +387,28 @@ ss::future<size_t> frontend::size_bytes() {
     co_return l0_size + l1_size_res.value().size;
 }
 
-std::unique_ptr<model::record_batch_reader::impl> frontend::make_l1_reader(
+model::record_batch_reader frontend::make_l1_reader(
   const cloud_topic_log_reader_config& cfg,
   model::topic_id_partition tidp) const {
     auto ct_state = _partition->get_cloud_topics_state();
+    auto* cache = ct_state->local().get_l1_reader_cache();
+    if (cache) {
+        if (auto hit = cache->get_reader(tidp, cfg); hit) {
+            return std::move(*hit);
+        }
+    }
+
     auto l1_metastore = ct_state->local().get_l1_metastore();
     auto l1_io = ct_state->local().get_l1_io();
     auto l1_reader_probe = ct_state->local().get_l1_reader_probe();
 
-    return std::make_unique<level_one_log_reader_impl>(
+    auto reader = std::make_unique<level_one_log_reader_impl>(
       cfg, _partition->ntp(), tidp, l1_metastore, l1_io, l1_reader_probe);
+
+    if (cache) {
+        return cache->put(std::move(reader));
+    }
+    return model::record_batch_reader(std::move(reader));
 }
 
 ss::future<std::optional<storage::timequery_result>>

src/v/cloud_topics/frontend/frontend.h

@@ -238,7 +238,7 @@ class frontend final {
     std::unique_ptr<model::record_batch_reader::impl>
     make_l0_reader(const cloud_topic_log_reader_config& cfg) const;
 
-    std::unique_ptr<model::record_batch_reader::impl> make_l1_reader(
+    model::record_batch_reader make_l1_reader(
       const cloud_topic_log_reader_config& cfg,
       model::topic_id_partition tidp) const;