ct/l1: add reader-keyed l1_reader_cache

Introduce a new l1_reader_cache modeled after storage::readers_cache.
Instead of caching open streams by object_id (the old approach deleted
in the previous commit), the new cache stores entire reader objects
keyed by topic_id_partition + offset.

The cache supports:
- get_reader: finds an idle reader matching tidp and offset, resets its
  config, and wraps it in a record_batch_reader with an RAII guard that
  returns it to the cache on destruction
- put: inserts a reusable reader into the idle pool
- TTL-based eviction (default 60s) and configurable size cap (default 128)
- Graceful shutdown that waits for in-use readers to return

New config properties:
- cloud_topics_l1_reader_cache_eviction_timeout_ms (default 60000)
- cloud_topics_l1_reader_cache_max_size (default 128, bounded 0-10000)

The cache is wired through state_accessors and constructed in
cloud_topics::app, but is not yet used in the fetch path. The reader
gains public methods (get_flags, reset_config, is_reusable,
close_stream_if_open, ntp, tidp) needed by the cache.

Author: wdberkeley

src/v/cloud_topics/BUILD

@@ -132,6 +132,7 @@ redpanda_cc_library(
         "//src/v/cloud_topics/level_one/common:file_io",
         "//src/v/cloud_topics/level_one/compaction:scheduler",
         "//src/v/cloud_topics/level_one/domain:domain_supervisor",
+        "//src/v/cloud_topics/level_one/frontend_reader:l1_reader_cache",
         "//src/v/cloud_topics/level_one/frontend_reader:level_one_reader_probe",
         "//src/v/cloud_topics/level_one/metastore:leader_router",
         "//src/v/cloud_topics/level_one/metastore:replicated_metastore",

src/v/cloud_topics/app.cc

@@ -26,6 +26,7 @@
 #include "cloud_topics/topic_manifest_upload_manager.h"
 #include "cluster/controller.h"
 #include "cluster/utils/partition_change_notifier_impl.h"
+#include "config/configuration.h"
 #include "config/node_config.h"
 #include "resource_mgmt/cpu_scheduling.h"
 #include "ssx/future-util.h"
@@ -70,6 +71,17 @@ ss::future<> app::construct(
 
     co_await construct_service(_l1_reader_probe);
 
+    co_await construct_service(
+      _l1_reader_cache,
+      ss::sharded_parameter([] {
+          return config::shard_local_cfg()
+            .cloud_topics_l1_reader_cache_eviction_timeout_ms.bind();
+      }),
+      ss::sharded_parameter([] {
+          return config::shard_local_cfg()
+            .cloud_topics_l1_reader_cache_max_size.bind();
+      }));
+
     co_await construct_service(
       l1_io,
       config::node().l1_staging_path(),
@@ -129,6 +141,7 @@ ss::future<> app::construct(
       ss::sharded_parameter(
         [&metadata_cache] { return &metadata_cache->local(); }),
       ss::sharded_parameter([this] { return &_l1_reader_probe.local(); }),
+      ss::sharded_parameter([this] { return &_l1_reader_cache.local(); }),
       ss::sharded_parameter([this] { return &rr_metadata_manager_.local(); }),
       ss::sharded_parameter([this] { return &rr_snapshot_manager_.local(); }));
 

src/v/cloud_topics/app.h

@@ -13,6 +13,7 @@
 #include "cloud_topics/level_one/common/file_io.h"
 #include "cloud_topics/level_one/compaction/scheduler.h"
 #include "cloud_topics/level_one/domain/domain_supervisor.h"
+#include "cloud_topics/level_one/frontend_reader/l1_reader_cache.h"
 #include "cloud_topics/level_one/frontend_reader/level_one_reader_probe.h"
 #include "cloud_topics/level_one/metastore/leader_router.h"
 #include "cloud_topics/level_one/metastore/replicated_metastore.h"
@@ -101,6 +102,7 @@ class app : public ssx::sharded_service_container {
 
     ss::sstring _logger_name;
     ss::sharded<level_one_reader_probe> _l1_reader_probe;
+    ss::sharded<l1_reader_cache> _l1_reader_cache;
     std::unique_ptr<data_plane_api> data_plane;
     ss::sharded<state_accessors> state;
     ss::sharded<l1::file_io> l1_io;

src/v/cloud_topics/level_one/frontend_reader/BUILD

@@ -16,6 +16,28 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "l1_reader_cache",
+    srcs = [
+        "l1_reader_cache.cc",
+    ],
+    hdrs = [
+        "l1_reader_cache.h",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":reader",
+        "//src/v/cloud_topics:log_reader_config",
+        "//src/v/cloud_topics:logger",
+        "//src/v/config",
+        "//src/v/container:intrusive",
+        "//src/v/model",
+        "//src/v/random:time_jitter",
+        "//src/v/ssx:future_util",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "reader",
     srcs = [

src/v/cloud_topics/level_one/frontend_reader/l1_reader_cache.cc

@@ -0,0 +1,226 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#include "cloud_topics/level_one/frontend_reader/l1_reader_cache.h"
+
+#include "cloud_topics/logger.h"
+#include "random/simple_time_jitter.h"
+#include "ssx/future-util.h"
+
+#include <seastar/core/coroutine.hh>
+#include <seastar/core/lowres_clock.hh>
+
+namespace cloud_topics {
+
+l1_reader_cache::l1_reader_cache(
+  config::binding<std::chrono::milliseconds> eviction_timeout,
+  config::binding<size_t> target_max_size)
+  : _eviction_timeout(std::move(eviction_timeout))
+  , _target_max_size(std::move(target_max_size)) {
+    _eviction_timer.set_callback([this] {
+        ssx::spawn_with_gate(_gate, [this] {
+            return maybe_evict().finally([this] { arm_eviction_timer(); });
+        });
+    });
+    arm_eviction_timer();
+}
+
+l1_reader_cache::~l1_reader_cache() {
+    vassert(
+      _readers.empty() && _in_use.empty(),
+      "l1_reader_cache must be stopped before destruction");
+}
+
+std::optional<model::record_batch_reader> l1_reader_cache::get_reader(
+  const model::topic_id_partition& tidp,
+  const cloud_topic_log_reader_config& cfg) {
+    if (_gate.is_closed()) {
+        return std::nullopt;
+    }
+
+    auto it = _readers.begin();
+    while (it != _readers.end()) {
+        if (
+          it->reader->is_reusable()
+          && it->reader->next_read_lower_bound() == cfg.start_offset
+          && it->reader->tidp() == tidp) {
+            break;
+        }
+        ++it;
+    }
+
+    if (it == _readers.end()) {
+        ++_cache_misses;
+        return std::nullopt;
+    }
+
+    auto& e = *it;
+    e.reader->reset_config(cfg);
+    ++_cache_hits;
+
+    _readers.erase(_readers.iterator_to(e));
+    _in_use.push_back(e);
+    return e.make_cached_reader(this);
+}
+
+model::record_batch_reader
+l1_reader_cache::put(std::unique_ptr<level_one_log_reader_impl> reader) {
+    if (_gate.is_closed()) {
+        return model::record_batch_reader(std::move(reader));
+    }
+
+    auto* ptr = new entry{.reader = std::move(reader)}; // NOLINT
+    _in_use.push_back(*ptr);
+    ++_readers_added;
+    maybe_evict_size();
+    return ptr->make_cached_reader(this);
+}
+
+l1_reader_cache::stats l1_reader_cache::get_stats() const {
+    return stats{
+      .in_use_readers = _in_use.size(),
+      .cached_readers = _readers.size(),
+    };
+}
+
+l1_reader_cache::entry_guard::~entry_guard() noexcept {
+    _cache->_in_use.erase(_cache->_in_use.iterator_to(*_e));
+    if (_e->reader->is_reusable()) {
+        _e->last_used = ss::lowres_clock::now();
+        _cache->_readers.push_back(*_e);
+    } else {
+        _cache->dispose_in_background(_e);
+    }
+    _cache->_in_use_reader_destroyed.broadcast();
+}
+
+model::record_batch_reader
+l1_reader_cache::entry::make_cached_reader(l1_reader_cache* cache) {
+    class cached_reader_impl final : public model::record_batch_reader::impl {
+    public:
+        explicit cached_reader_impl(entry* e, l1_reader_cache* c)
+          : _underlying(e->reader.get())
+          , _guard(e, c) {}
+        cached_reader_impl(cached_reader_impl&&) noexcept = default;
+        cached_reader_impl& operator=(cached_reader_impl&&) noexcept = default;
+        cached_reader_impl(const cached_reader_impl&) = delete;
+        cached_reader_impl& operator=(const cached_reader_impl&) = delete;
+
+        bool is_end_of_stream() const final {
+            return _underlying->is_end_of_stream();
+        }
+
+        ss::future<model::record_batch_reader::storage_t>
+        do_load_slice(model::timeout_clock::time_point tout) final {
+            return _underlying->do_load_slice(tout);
+        }
+
+        std::optional<private_flags> get_flags() const final {
+            return _underlying->get_flags();
+        }
+
+        ss::future<> finally() noexcept final { return ss::now(); }
+
+        fmt::iterator format_to(fmt::iterator it) const final {
+            return _underlying->format_to(it);
+        }
+        ~cached_reader_impl() final = default;
+
+    private:
+        level_one_log_reader_impl* _underlying;
+        entry_guard _guard;
+    };
+
+    return model::make_record_batch_reader<cached_reader_impl>(this, cache);
+}
+
+ss::future<> l1_reader_cache::wait_for_no_inuse_readers() {
+    return _in_use_reader_destroyed.wait([this] { return _in_use.empty(); });
+}
+
+ss::future<> l1_reader_cache::stop() {
+    if (_eviction_timer.armed()) {
+        _eviction_timer.cancel();
+    }
+    co_await _gate.close();
+    co_await wait_for_no_inuse_readers();
+
+    // Close all cached readers' streams.
+    for (auto& r : _readers) {
+        co_await r.reader->finally();
+    }
+    _readers.clear_and_dispose([](entry* e) {
+        delete e; // NOLINT
+    });
+}
+
+void l1_reader_cache::dispose_in_background(entry* e) {
+    if (_gate.is_closed()) {
+        _readers.push_back(*e);
+    } else {
+        ++_readers_evicted;
+        ssx::spawn_with_gate(_gate, [e] {
+            return e->reader->finally().finally([e] {
+                delete e; // NOLINT
+            });
+        });
+    }
+}
+
+void l1_reader_cache::arm_eviction_timer() {
+    if (_gate.is_closed()) {
+        return;
+    }
+    auto timeout = _eviction_timeout();
+    if (timeout > std::chrono::milliseconds::zero()) {
+        _eviction_timer.arm(
+          simple_time_jitter<ss::lowres_clock>(timeout).next_duration());
+    }
+}
+
+ss::future<> l1_reader_cache::maybe_evict() {
+    auto now = ss::lowres_clock::now();
+    uncounted_intrusive_list<entry, &entry::_hook> to_evict;
+
+    for (auto it = _readers.begin(); it != _readers.end();) {
+        const auto invalid = !it->reader->is_reusable();
+        const auto outdated = it->last_used + _eviction_timeout() < now;
+        if (invalid || outdated) {
+            it = _readers.erase_and_dispose(
+              it, [&to_evict](entry* e) { to_evict.push_back(*e); });
+        } else {
+            ++it;
+        }
+    }
+
+    for (auto it = to_evict.begin(); it != to_evict.end();) {
+        auto* e = &*it;
+        ++it;
+        to_evict.erase(to_evict.iterator_to(*e));
+        co_await e->reader->finally();
+        ++_readers_evicted;
+        delete e; // NOLINT
+    }
+}
+
+bool l1_reader_cache::over_size_limit() const {
+    return !_readers.empty()
+           && _readers.size() + _in_use.size() > _target_max_size();
+}
+
+void l1_reader_cache::maybe_evict_size() {
+    if (!over_size_limit()) [[likely]] {
+        return;
+    }
+    _readers.pop_front_and_dispose(
+      [this](entry* e) { dispose_in_background(e); });
+}
+
+} // namespace cloud_topics