storage: admit tiered_cloud partitions to space management

src/v/cloud_topics/level_zero/stm/ctp_stm.cc

@@ -116,8 +116,9 @@ ss::future<> ctp_stm::prefix_truncate_bg() {
         // trims aggressively (the local log only holds placeholders);
         // storage.mode=tiered_cloud honors local retention plus the compaction
         // floor and so keeps more data locally.
+        const bool is_tiered = _raft->log()->config().is_tiered_cloud();
         model::offset target;
-        if (_raft->log()->config().is_tiered_cloud()) {
+        if (is_tiered) {
             target = co_await compute_local_retention_offset();
         } else {
             // storage.mode=cloud keeps only placeholders locally; the
@@ -133,10 +134,12 @@ ss::future<> ctp_stm::prefix_truncate_bg() {
           _raft->last_snapshot_index());
         try {
             if (
-              _raft->last_snapshot_index() >= target
-              && _active_readers.empty()) {
+              _raft->last_snapshot_index() >= target && _active_readers.empty()
+              && !is_tiered) {
                 // Only wait without a timeout if there are no active readers
-                // that could be holding us back.
+                // that could be holding us back. tiered_cloud always polls
+                // because its space-management offset is set without signalling
+                // _lro_advanced.
                 co_await _lro_advanced.wait();
             } else {
                 co_await _lro_advanced.wait(retry_backoff_time);

src/v/resource_mgmt/storage.cc

@@ -289,7 +289,9 @@ eviction_policy::collect_reclaimable_offsets() {
      */
     chunked_vector<ss::lw_shared_ptr<cluster::partition>> partitions;
     for (const auto& p : _pm->local().partitions()) {
-        if (!p.second->remote_partition()) {
+        if (
+          !p.second->remote_partition()
+          && !p.second->log()->config().is_tiered_cloud()) {
             continue;
         }
         partitions.push_back(p.second);

src/v/storage/disk_log_impl.cc

@@ -1228,9 +1228,9 @@ disk_log_impl::maybe_apply_local_storage_overrides(gc_config cfg) const {
 
     // cloud_retention is disabled, do not override. Cloud-topic partitions
     // (storage.mode in {cloud, tiered_cloud}) bypass this gate:
-    // is_cloud_retention_active() is false for them, but ctp_stm still needs
+    // is_archival_active() is false for them, but ctp_stm still needs
     // the local-target override to engage under retention_local_strict.
-    if (!is_cloud_retention_active() && !config().cloud_topic_enabled()) {
+    if (!is_archival_active() && !config().cloud_topic_enabled()) {
         return cfg;
     }
 
@@ -1310,11 +1310,15 @@ gc_config disk_log_impl::apply_local_storage_overrides(gc_config cfg) const {
     return cfg;
 }
 
-bool disk_log_impl::is_cloud_retention_active() const {
+bool disk_log_impl::is_archival_active() const {
     return config::shard_local_cfg().cloud_storage_enabled()
            && (config().is_archival_enabled());
 }
 
+bool disk_log_impl::is_cloud_gc_active() const {
+    return is_archival_active() || config().is_tiered_cloud();
+}
+
 /*
  * applies overrides for non-cloud storage settings
  */
@@ -1358,6 +1362,11 @@ gc_config disk_log_impl::apply_overrides(gc_config defaults) const {
 }
 
 ss::future<> disk_log_impl::housekeeping(housekeeping_config cfg) {
+    // Cloud topics do local retention/compaction elsewhere (ctp_stm trims the
+    // local log; compaction is at L1), so this loop does nothing for them.
+    if (config().cloud_topic_enabled()) {
+        co_return;
+    }
     ss::gate::holder holder{_compaction_housekeeping_gate};
     vlog(
       gclog.trace,
@@ -1763,12 +1772,21 @@ ss::future<> disk_log_impl::rewrite_segment_with_offset_map(
 }
 
 ss::future<> disk_log_impl::gc(gc_config cfg) {
+    // Cloud topics do local retention/compaction elsewhere (ctp_stm trims the
+    // local log; compaction is at L1), so disk_log_impl gc does not apply.
+    if (config().cloud_topic_enabled()) {
+        co_return;
+    }
     ss::gate::holder holder{_compaction_housekeeping_gate};
     co_await do_gc(cfg);
 }
 
 ss::future<std::optional<model::offset>> disk_log_impl::do_gc(gc_config cfg) {
     vassert(!_closed, "gc on closed log - {}", *this);
+    vassert(
+      !config().cloud_topic_enabled(),
+      "[{}] gc on cloud topic partition",
+      config().ntp());
 
     cfg = apply_overrides(cfg);
 
@@ -1783,7 +1801,7 @@ ss::future<std::optional<model::offset>> disk_log_impl::do_gc(gc_config cfg) {
         const auto offset = _cloud_gc_offset.value();
         _cloud_gc_offset.reset();
 
-        if (!is_cloud_retention_active()) {
+        if (!is_archival_active()) {
             vlog(
               gclog.warn,
               "[{}] expected remote retention to be active",
@@ -1916,12 +1934,12 @@ disk_log_impl::compute_gc_offset(gc_config cfg) {
     // ctp_stm for cloud-topic partitions. The offset is retention-driven
     // unless space management has pinned _cloud_gc_offset, which then takes
     // precedence. maybe_apply_local_storage_overrides
-    // bypasses the is_cloud_retention_active() gate for cloud-topic partitions
+    // bypasses the is_archival_active() gate for cloud-topic partitions
     // (the gate is false for storage.mode in {cloud, tiered_cloud}), so the
     // local-target override engages there under retention_local_strict.
     cfg = apply_kafka_retention_overrides(cfg);
     if (_cloud_gc_offset.has_value()) {
-        co_return _cloud_gc_offset;
+        co_return std::exchange(_cloud_gc_offset, std::nullopt);
     }
     co_await maybe_adjust_retention_timestamps();
     cfg = maybe_apply_local_storage_overrides(cfg);
@@ -3921,7 +3939,7 @@ disk_log_impl::disk_usage_and_reclaimable_space(gc_config input_cfg) {
           && seg->offsets().get_dirty_offset() <= retention_offset.value()) {
             retention_segments.push_back(seg);
         } else if (
-          is_cloud_retention_active()
+          is_cloud_gc_active()
           && seg->offsets().get_dirty_offset() <= max_removable) {
             available_segments.push_back(seg);
         } else {
@@ -3937,8 +3955,8 @@ disk_log_impl::disk_usage_and_reclaimable_space(gc_config input_cfg) {
          * get_reclaimable_offsets is going to be merged together.
          */
         if (
-          !config().is_read_replica_mode_enabled()
-          && is_cloud_retention_active() && seg != _segs.back()
+          !config().is_read_replica_mode_enabled() && is_cloud_gc_active()
+          && seg != _segs.back()
           && seg->offsets().get_dirty_offset() <= max_removable
           && local_retention_offset.has_value()
           && seg->offsets().get_dirty_offset()
@@ -4279,8 +4297,8 @@ ss::future<usage_report> disk_log_impl::disk_usage(gc_config cfg) {
 chunked_vector<ss::lw_shared_ptr<segment>>
 disk_log_impl::cloud_gc_eligible_segments() {
     vassert(
-      is_cloud_retention_active(),
-      "Expected {} to have cloud retention enabled",
+      is_cloud_gc_active(),
+      "Expected cloud GC to be active for {}",
       config().ntp());
 
     constexpr size_t keep_segs = 1;
@@ -4313,7 +4331,7 @@ disk_log_impl::cloud_gc_eligible_segments() {
 }
 
 void disk_log_impl::set_cloud_gc_offset(model::offset offset) {
-    if (!is_cloud_retention_active()) {
+    if (!is_cloud_gc_active()) {
         vlog(
           stlog.debug,
           "Ignoring request to set GC offset on non-cloud enabled partition "
@@ -4338,7 +4356,7 @@ disk_log_impl::get_reclaimable_offsets(gc_config cfg) {
 
     reclaimable_offsets res;
 
-    if (!is_cloud_retention_active()) {
+    if (!is_cloud_gc_active()) {
         vlog(
           stlog.debug,
           "Reporting no reclaimable offsets for non-cloud partition {}",
@@ -4540,7 +4558,7 @@ size_t disk_log_impl::reclaimable_size_bytes() const {
      * local retention size may change. catch these before reporting potentially
      * stale information.
      */
-    if (!is_cloud_retention_active()) {
+    if (!is_cloud_gc_active()) {
         return 0;
     }
     if (config().is_read_replica_mode_enabled()) {

src/v/storage/disk_log_impl.h

@@ -210,9 +210,9 @@ class disk_log_impl final : public log {
     /// Applies retention overrides (callers need not pre-apply them) and
     /// adjusts bogus (future) retention timestamps, mutating segment indexes
     /// when an entire segment is bogus, then returns the offset GC would
-    /// evict to -- usually derived from local retention, but when space
-    /// management has set _cloud_gc_offset that offset is returned instead
-    /// (without resetting it -- that is do_gc's responsibility).
+    /// evict to. Usually derived from local retention; when space management
+    /// has set _cloud_gc_offset, that offset is returned and consumed (reset)
+    /// here, so the caller is responsible for acting on it.
     ss::future<std::optional<model::offset>>
     compute_gc_offset(gc_config cfg) final;
 
@@ -421,7 +421,11 @@ class disk_log_impl final : public log {
     gc_config maybe_apply_local_storage_overrides(gc_config) const;
     gc_config apply_local_storage_overrides(gc_config) const;
 
-    bool is_cloud_retention_active() const;
+    bool is_archival_active() const;
+    // True when local segments are a reclaimable cache of cloud-resident data
+    // (legacy tiered storage or tiered_cloud); broader than
+    // is_archival_active(), which is archival-only.
+    bool is_cloud_gc_active() const;
 
     // returns retention_offset(cfg) but may also first apply adjustments to
     // future timestamps if this option is turned on in configuration.

tests/rptest/tests/tiered_cloud_local_retention_test.py

@@ -299,6 +299,138 @@ def test_local_data_grows_in_tiered_cloud(self):
             f"expected near {replication * self.local_target_bytes} bytes"
         )
 
+    @cluster(num_nodes=4)
+    def test_capacity_pressure_reclaims_tiered_cloud(self):
+        """
+        Cluster-wide disk pressure (the disk_space_manager capacity path)
+        reclaims a tiered_cloud partition's local log, not just the
+        retention.local.target path.
+
+        The topic sets no local target and effectively infinite retention, so
+        only the capacity path can bound it. On a build that excludes
+        tiered_cloud partitions from disk_space_manager the local log grows
+        unbounded and this fails; with them admitted, the eviction policy pins
+        _cloud_gc_offset and ctp_stm trims to it, so the footprint converges
+        near the capacity target.
+        """
+        assert self.redpanda is not None
+
+        # Node-wide log-data target, small enough that the produced topic must
+        # be trimmed to meet it. disk_reservation_percent=0 and percent=100 so
+        # the bytes target is the effective one.
+        capacity_target = 16 * 1024 * 1024
+        self.redpanda.set_cluster_config(
+            {
+                "retention_local_target_capacity_bytes": capacity_target,
+                "retention_local_target_capacity_percent": 100,
+                "disk_reservation_percent": 0,
+            }
+        )
+
+        # No retention.local.target.bytes: the local-target path stays inert
+        # (only the 1-day default applies, and nothing is that old), so only
+        # the capacity path can bound this topic. retention.bytes is
+        # effectively unlimited, mirroring the field scenario that filled the
+        # disk.
+        rpk = RpkTool(self.redpanda)
+        rpk.create_topic(
+            topic=self.topic_name,
+            partitions=1,
+            replicas=3,
+            config={
+                TopicSpec.PROPERTY_STORAGE_MODE: (TopicSpec.STORAGE_MODE_TIERED_CLOUD),
+                "cleanup.policy": TopicSpec.CLEANUP_DELETE,
+                "retention.bytes": str(self.retention_bytes),
+                "segment.bytes": str(self.segment_size),
+            },
+        )
+        self._wait_for_partition_info()
+
+        # Produce well past the capacity target so an untrimmed build grows far
+        # beyond it.
+        produce_bytes = 64 * 1024 * 1024
+        self._produce(produce_bytes)
+        self.wait_until_reconciled(topic=self.topic_name, partition=0)
+
+        # The target is per node; the metric sums this topic's 3 replicas and
+        # excludes internal logs (topic-filtered). Allow 3x the target plus
+        # per-replica headroom for the never-evicted active segment and
+        # trim-cycle latency. Still far below the ~3x produce an untrimmed
+        # build retains.
+        replication = 3
+        ceiling = replication * (capacity_target + 4 * self.segment_size)
+        # The topic has no local retention target and a lenient retention.bytes,
+        # so ctp_stm's own retention never trims at this volume; only a
+        # disk_space_manager capacity pin can drive the reclaim. Staying below
+        # the ceiling therefore proves the capacity path engaged for a
+        # tiered_cloud partition.
+        self._wait_local_below(ceiling_bytes=ceiling, timeout_sec=180)
+
+    @cluster(num_nodes=4)
+    def test_idle_partition_reclaims_on_pin(self):
+        """
+        A tiered_cloud partition that has gone idle still reclaims when disk
+        pressure sets _cloud_gc_offset afterwards.
+
+        With no produce the LRO is quiescent, so nothing signals ctp_stm's
+        truncate loop. It wakes on its bounded poll interval, re-reads the
+        offset disk_space_manager set, and trims to it, so an idle partition
+        reclaims without further produce.
+        """
+        assert self.redpanda is not None
+
+        capacity_target = 16 * 1024 * 1024
+        replication = 3
+        ceiling = replication * (capacity_target + 4 * self.segment_size)
+
+        # Capacity path effectively off: percent=100 with no bytes target lets
+        # the backlog accumulate without pressure, so nothing trims until the
+        # bytes target is set below.
+        self.redpanda.set_cluster_config(
+            {
+                "retention_local_target_capacity_percent": 100,
+                "disk_reservation_percent": 0,
+            }
+        )
+
+        rpk = RpkTool(self.redpanda)
+        rpk.create_topic(
+            topic=self.topic_name,
+            partitions=1,
+            replicas=3,
+            config={
+                TopicSpec.PROPERTY_STORAGE_MODE: (TopicSpec.STORAGE_MODE_TIERED_CLOUD),
+                "cleanup.policy": TopicSpec.CLEANUP_DELETE,
+                "retention.bytes": str(self.retention_bytes),
+                "segment.bytes": str(self.segment_size),
+            },
+        )
+        self._wait_for_partition_info()
+
+        # Fill the local log, then let reconciliation catch up so the LRO is
+        # quiescent (nothing to drive the truncate loop). Nothing trims yet
+        # (no pressure, lenient retention.bytes), so the footprint holds near
+        # the produced size.
+        self._produce(64 * 1024 * 1024)
+        self.wait_until_reconciled(topic=self.topic_name, partition=0)
+
+        # Confirm the backlog is present and untrimmed before applying pressure,
+        # so the reclaim assertion below cannot pass trivially.
+        wait_until(
+            lambda: self._local_partition_bytes() > ceiling,
+            timeout_sec=60,
+            backoff_sec=2,
+            err_msg="backlog never reached the expected size before pressure",
+        )
+
+        # Apply pressure: disk_space_manager sets _cloud_gc_offset on its next
+        # cycle. The LRO is quiescent, so the truncate loop only observes it on
+        # its next bounded poll, then trims to it.
+        self.redpanda.set_cluster_config(
+            {"retention_local_target_capacity_bytes": capacity_target}
+        )
+        self._wait_local_below(ceiling_bytes=ceiling, timeout_sec=120)
+
     @cluster(num_nodes=4)
     def test_compact_topic_evicts_via_l1_floor(self):
         """