storage: include tiered_cloud partitions in disk space management

tiered_cloud partition local logs were excluded from the disk space
manager, so they were never reclaimed under disk pressure. A new
is_cloud_gc_active() predicate (archival retention or tiered_cloud) gates
the space manager's reclaim paths, so tiered_cloud partitions are now
collected below the reconciled L1 offset, like legacy tiered storage.

For these partitions ctp_stm is the reclaim executor: do_gc leaves the
_cloud_gc_offset pin for it, and compute_gc_offset consumes the pin and
folds it with retention.

Author: oleiman

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

@@ -1315,6 +1315,10 @@ bool disk_log_impl::is_cloud_retention_active() const {
            && (config().is_archival_enabled());
 }
 
+bool disk_log_impl::is_cloud_gc_active() const {
+    return is_cloud_retention_active() || config().is_tiered_cloud();
+}
+
 /*
  * applies overrides for non-cloud storage settings
  */
@@ -1779,7 +1783,11 @@ ss::future<std::optional<model::offset>> disk_log_impl::do_gc(gc_config cfg) {
      * process such as disk space management drives this process such that after
      * a round of gc has run the intent flag can be cleared.
      */
-    if (_cloud_gc_offset.has_value()) {
+    // Cloud-topic (tiered_cloud) partitions do not evict through do_gc /
+    // request_eviction_until_offset; ctp_stm consumes _cloud_gc_offset via
+    // compute_gc_offset instead. Leave the pin alone for them so do_gc does not
+    // steal and discard it before ctp_stm sees it.
+    if (_cloud_gc_offset.has_value() && !config().is_tiered_cloud()) {
         const auto offset = _cloud_gc_offset.value();
         _cloud_gc_offset.reset();
 
@@ -1912,20 +1920,29 @@ disk_log_impl::maybe_adjusted_retention_offset(gc_config cfg) {
 
 ss::future<std::optional<model::offset>>
 disk_log_impl::compute_gc_offset(gc_config cfg) {
-    // Single GC-offset computation shared by gc() housekeeping and by
-    // 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
-    // (the gate is false for storage.mode in {cloud, tiered_cloud}), so the
-    // local-target override engages there under retention_local_strict.
+    // Single GC-offset computation shared by gc() housekeeping and by ctp_stm
+    // for cloud-topic partitions. Retention (time/bytes plus the local-storage
+    // overrides, which maybe_apply_local_storage_overrides engages for
+    // cloud-topic partitions under retention_local_strict even though the
+    // is_cloud_retention_active() gate is false) sets the base offset. Space
+    // management can raise it under disk pressure via a pinned
+    // _cloud_gc_offset.
+    //
+    // ctp_stm is the sole consumer of the pin for cloud topics (do_gc leaves it
+    // alone), so consume it here instead of letting it persist: a stale pin
+    // would otherwise keep shadowing retention after disk pressure relents.
+    // Fold it with retention so the pin only ever raises the trim point.
     cfg = apply_kafka_retention_overrides(cfg);
-    if (_cloud_gc_offset.has_value()) {
-        co_return _cloud_gc_offset;
-    }
     co_await maybe_adjust_retention_timestamps();
     cfg = maybe_apply_local_storage_overrides(cfg);
-    co_return retention_offset(cfg);
+    auto retention = retention_offset(cfg);
+
+    auto pin = _cloud_gc_offset;
+    _cloud_gc_offset.reset();
+    if (pin.has_value()) {
+        co_return retention.has_value() ? std::max(*retention, *pin) : pin;
+    }
+    co_return retention;
 }
 
 std::optional<model::offset>
@@ -3921,7 +3938,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 +3954,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 +4296,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 +4330,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 +4355,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 +4557,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,10 @@ 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 retention and local-storage overrides;
+    /// when space management has pinned _cloud_gc_offset it is folded in (max)
+    /// so the pin only raises the trim point. The pin is consumed (reset) here;
+    /// for tiered_cloud, ctp_stm is its sole consumer (do_gc leaves it alone).
     ss::future<std::optional<model::offset>>
     compute_gc_offset(gc_config cfg) final;
 
@@ -422,6 +423,10 @@ class disk_log_impl final : public log {
     gc_config apply_local_storage_overrides(gc_config) const;
 
     bool is_cloud_retention_active() const;
+    // True when local segments are a reclaimable cache of cloud-resident data
+    // (legacy tiered storage or tiered_cloud); broader than
+    // is_cloud_retention_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.