RFC ts-ct-migration: partition_mode substrate [PR 0]

src/v/cluster/BUILD

@@ -1407,6 +1407,7 @@ redpanda_cc_library(
         ":types",
         "//src/v/base",
         "//src/v/storage",
+        "//src/v/utils:prefix_logger",
         "@fmt",
         "@seastar",
     ],

src/v/cluster/archival/archival_metadata_stm.cc

@@ -1674,6 +1674,11 @@ archival_metadata_stm::manifest() const {
     return *_manifest;
 }
 
+bool archival_metadata_stm::holds_archived_data() const {
+    return _manifest->size() > 0
+           || _manifest->get_archive_start_offset() != model::offset{};
+}
+
 model::offset archival_metadata_stm::get_start_offset() const {
     auto p = _manifest->get_start_offset();
     if (p.has_value()) {

src/v/cluster/archival/archival_metadata_stm.h

@@ -216,6 +216,11 @@ class archival_metadata_stm final : public raft::persisted_stm<> {
     /// added with `add_segments`.
     const cloud_storage::partition_manifest& manifest() const;
 
+    /// True if the partition holds tiered-storage data: the manifest has
+    /// segments, or a non-empty spillover archive. Used to detect that a
+    /// partition is (still) tiered storage independent of the topic config.
+    bool holds_archived_data() const;
+
     ss::future<> stop() override;
 
     static ss::future<> make_snapshot(

src/v/cluster/partition.cc

@@ -473,6 +473,17 @@ ss::future<> partition::start(
     _partition_properties_stm
       = _raft->stm_manager()->get<cluster::partition_properties_stm>();
 
+    // Feed the partition's durable storage mode into the log's ntp_config and
+    // keep it up to date as the STM applies changes. Reading replicated STM
+    // state is always safe; writes (which require the partition_mode feature)
+    // are gated separately. Until partition_mode is set, partition_mode() is
+    // `unset` and ntp_config falls back to the topic-config-derived mode.
+    if (_partition_properties_stm) {
+        _partition_properties_stm->set_partition_mode_change_callback(
+          [this] { update_partition_mode(); });
+        update_partition_mode();
+    }
+
     // Start the probe after the partition is fully initialised
     _probe.setup_metrics(ntp);
 
@@ -1752,6 +1763,66 @@ partition::force_abort_replica_set_update(model::revision_id rev) {
 }
 consensus_ptr partition::raft() const { return _raft; }
 
+void partition::update_partition_mode() {
+    if (!_partition_properties_stm) {
+        return;
+    }
+    _raft->log()->set_partition_mode(
+      _partition_properties_stm->partition_mode());
+}
+
+ss::future<> partition::maybe_bootstrap_partition_mode() {
+    if (!_partition_properties_stm || !is_leader()) {
+        co_return;
+    }
+    if (!_feature_table.local().is_active(
+          features::feature::tiered_to_cloud_migration)) {
+        co_return;
+    }
+    // Already bootstrapped (or already advanced by a migration): nothing to do.
+    if (
+      _partition_properties_stm->partition_mode()
+      != model::redpanda_storage_mode::unset) {
+        co_return;
+    }
+    // Pick the mode to record.
+    //   - holds_archived_data(): the partition holds tiered-storage data, so it
+    //     is (still) tiered regardless of topic_mode -- covers a node that
+    //     became leader *after* the operator switched the topic ts->ct, whose
+    //     archival manifest (loaded on a running partition) reveals it is
+    //     mid-migration while topic_mode() already reads `cloud`.
+    //   - _creation_partition_mode: the classification manage() computed
+    //     deterministically at creation, authoritative when set. Preferred over
+    //     re-reading topic_mode() here because it captures the mode as it was
+    //     at creation, immune to a later async topic-config flip.
+    //   - topic_mode(): the fallback for a partition with no creation-time hint
+    //     (created before this classifier existed). Legacy shadow_indexing
+    //     topics have topic_mode() == unset and are left unset (SI fallback).
+    const bool holds_ts_data = _archival_meta_stm
+                               && _archival_meta_stm->holds_archived_data();
+    model::redpanda_storage_mode target;
+    if (holds_ts_data) {
+        target = model::redpanda_storage_mode::tiered;
+    } else if (
+      _creation_partition_mode != model::redpanda_storage_mode::unset) {
+        target = _creation_partition_mode;
+    } else {
+        target = get_ntp_config().topic_mode();
+    }
+    if (target == model::redpanda_storage_mode::unset) {
+        co_return;
+    }
+    auto res = co_await _partition_properties_stm->set_partition_mode(target);
+    if (res.has_error()) {
+        vlog(
+          clusterlog.debug,
+          "{}: failed to bootstrap partition_mode to {}: {}",
+          _raft->ntp(),
+          target,
+          res.error().message());
+    }
+}
+
 ss::future<result<model::offset>> partition::set_writes_disabled(
   partition_properties_stm::writes_disabled disable,
   model::timeout_clock::time_point deadline,

src/v/cluster/partition.h

@@ -164,6 +164,25 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
 
     bool is_elected_leader() const;
     bool is_leader() const;
+
+    // On becoming leader, capture the partition's current
+    // (topic-config-derived) storage mode into partition_properties, so it is
+    // remembered across later topic-config changes (the TS->CT migration).
+    // Gated behind the partition_mode feature; one-time per partition and
+    // idempotent. Legacy shadow_indexing topics (no explicit storage_mode) are
+    // left unset. Invoked by partition_manager's leadership notification.
+    ss::future<> maybe_bootstrap_partition_mode();
+
+    // The partition_mode to record at bootstrap, classified deterministically
+    // at creation (partition_manager::manage), so it does not depend on the
+    // async config propagation a runtime re-derivation would race. `unset`
+    // means "no creation-time classification" (e.g. created while the feature
+    // was inactive), in which case the bootstrap falls back to the runtime
+    // signals.
+    void set_creation_partition_mode(model::redpanda_storage_mode m) {
+        _creation_partition_mode = m;
+    }
+
     bool has_followers() const;
     void block_new_leadership() const;
     void unblock_new_leadership() const;
@@ -424,6 +443,16 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
     // dirty so that it gets reuploaded
     ss::future<> restart_archiver(bool should_notify_topic_config);
 
+    // Push the partition's durable storage mode (partition_properties_stm) into
+    // the log's ntp_config, so ntp_config::partition_mode() reflects it. Called
+    // at start and whenever the STM signals a change.
+    void update_partition_mode();
+
+    // Creation-time partition_mode classification; see
+    // set_creation_partition_mode. `unset` until manage() sets it.
+    model::redpanda_storage_mode _creation_partition_mode{
+      model::redpanda_storage_mode::unset};
+
     consensus_ptr _raft; // never null
     ss::shared_ptr<cluster::log_eviction_stm> _log_eviction_stm;
     ss::shared_ptr<cluster::rm_stm> _rm_stm;

src/v/cluster/partition_manager.cc

@@ -25,11 +25,13 @@
 #include "cluster/partition_recovery_manager.h"
 #include "cluster/topic_configuration.h"
 #include "cluster/types.h"
+#include "features/feature_table.h"
 #include "model/metadata.h"
 #include "raft/consensus.h"
 #include "raft/consensus_utils.h"
 #include "raft/fundamental.h"
 #include "ssx/async-clear.h"
+#include "ssx/future-util.h"
 
 #include <seastar/core/lowres_clock.hh>
 #include <seastar/core/shared_ptr.hh>
@@ -73,6 +75,11 @@ partition_manager::partition_manager(
                 if (a) {
                     a.value().get().notify_leadership(leader_id);
                 }
+                // Bootstrap the partition's durable storage mode on becoming
+                // leader (one-time, idempotent; no-op when not leader). The
+                // shared_ptr keeps the partition alive across the async write.
+                ssx::spawn_with_gate(
+                  _gate, [p] { return p->maybe_bootstrap_partition_mode(); });
             }
         });
     _shutdown_watchdog.set_callback(
@@ -100,9 +107,36 @@ partition_manager::get_topic_partition_table(
 
 ss::future<> partition_manager::start() {
     maybe_arm_shutdown_watchdog();
+    // Bootstrap partition_mode once the migration feature activates. The
+    // leadership-notification bootstrap (maybe_bootstrap_partition_mode) is
+    // gated on the feature being active, so a partition that became leader
+    // while the feature was inactive is left with partition_mode == unset.
+    // Re-run the (idempotent) bootstrap on every current leader when the
+    // feature turns active; partitions that gain leadership after activation
+    // are covered by the leadership notification.
+    ssx::spawn_with_gate(_gate, [this] {
+        return bootstrap_partition_mode_on_migration_feature();
+    });
     co_return;
 }
 
+ss::future<>
+partition_manager::bootstrap_partition_mode_on_migration_feature() {
+    try {
+        co_await _feature_table.local().await_feature(
+          features::feature::tiered_to_cloud_migration, _as);
+    } catch (...) {
+        // Aborted on shutdown before the feature activated.
+        co_return;
+    }
+    for (auto& [_, p] : _ntp_table) {
+        if (p->is_leader()) {
+            ssx::spawn_with_gate(
+              _gate, [p] { return p->maybe_bootstrap_partition_mode(); });
+        }
+    }
+}
+
 ss::future<consensus_ptr> partition_manager::manage(
   storage::ntp_config ntp_cfg,
   raft::group_id group,
@@ -330,6 +364,17 @@ ss::future<consensus_ptr> partition_manager::manage(
       read_replica_bucket,
       _cloud_topics_state);
 
+    // Populate partition_mode at creation when the feature is active, so it is
+    // classified deterministically here from the topic config rather than via
+    // the racy runtime bootstrap. The bootstrap remains for the feature-switch
+    // case (a partition that predates the feature activation). Left unset when
+    // the feature is inactive.
+    if (
+      _feature_table.local().is_active(
+        features::feature::tiered_to_cloud_migration)) {
+        p->set_creation_partition_mode(log->config().topic_mode());
+    }
+
     _ntp_table.emplace(log->config().ntp(), p);
     _raft_table.emplace(group, p);
 

src/v/cluster/partition_manager.h

@@ -277,6 +277,11 @@ class partition_manager
     void check_partitions_shutdown_state();
 
     void maybe_arm_shutdown_watchdog();
+
+    // Awaits the tiered_to_cloud_migration feature and then re-runs
+    // maybe_bootstrap_partition_mode on every current leader, covering
+    // partitions that became leader while the feature was inactive.
+    ss::future<> bootstrap_partition_mode_on_migration_feature();
     storage::api& _storage;
     /// used to wait for concurrent recoveries
     ss::sharded<raft::group_manager>& _raft_manager;