cluster/partition: bootstrap partition_mode on leadership and creation

Record the partition's durable storage mode (partition_mode) in
partition_properties. It is populated:
  - at partition creation, from the topic-config mode, when the migration
    feature is active (partition_manager::manage);
  - on becoming leader, for a partition that predates this (maybe_bootstrap_
    partition_mode); a node that became leader after an in-place ts->ct switch
    keys on the archival manifest (holds_archived_data) since topic_mode()
    already reads cloud;
  - on migration-feature activation, sweeping current leaders that were elected
    while the feature was inactive.
The creation-time classification (set_creation_partition_mode) is preferred by
the bootstrap over re-reading topic_mode(), so the mode does not depend on the
async config propagation a runtime re-derivation would race.

Also add partition_properties_stm::seed_partition_mode(), which pre-writes the
kvstore-backed local snapshot before the STM starts so a partition restored with
a non-empty log comes up with partition_mode already set (the seed survives
start(), whose initial cleanup only runs for an empty log). Used by the
migration-recovery path.

partition_mode stays inert here; serving/behavior accessors begin keying on it
in later PRs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: sjust-redpanda

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

@@ -1771,6 +1771,58 @@ void partition::update_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;
@@ -429,6 +448,11 @@ class partition : public ss::enable_lw_shared_from_this<partition> {
     // 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;

src/v/cluster/partition_properties_stm.cc

@@ -21,6 +21,7 @@
 #include "model/record_batch_types.h"
 #include "raft/persisted_stm.h"
 #include "storage/log.h"
+#include "utils/prefix_logger.h"
 
 #include <fmt/format.h>
 
@@ -143,6 +144,30 @@ ss::future<raft::stm_snapshot> partition_properties_stm::take_local_snapshot(
       serde::to_iobuf(local_snapshot{.state_updates = std::move(state)}));
 }
 
+ss::future<> partition_properties_stm::seed_partition_mode(
+  storage::kvstore& kvstore,
+  ss::logger& logger,
+  const model::ntp& ntp,
+  model::offset snapshot_offset,
+  model::redpanda_storage_mode mode) {
+    // A single base state entry (update_offset {}, strictly below any log start
+    // offset) carrying the mode, matching the STM's initial in-memory state
+    // plus partition_mode.
+    local_snapshot snap;
+    snap.state_updates.push_back(
+      state_snapshot{
+        .writes_disabled = writes_disabled::no,
+        .update_offset = model::offset{},
+        .writes_revision_id = {},
+        .partition_mode = mode});
+    prefix_logger log(logger, fmt::format("[{}]", ntp));
+    raft::kvstore_backed_stm_snapshot backend(
+      partition_properties_stm_snapshot, log, ntp, kvstore);
+    co_await backend.persist_local_snapshot(
+      raft::stm_snapshot::create(
+        0, snapshot_offset, serde::to_iobuf(std::move(snap))));
+}
+
 ss::future<> partition_properties_stm::do_apply(const model::record_batch& b) {
     if (
       b.header().type

src/v/cluster/partition_properties_stm.h

@@ -65,6 +65,21 @@ class partition_properties_stm
     // callers (ntp_config) fall back to the topic-config-derived mode.
     model::redpanda_storage_mode partition_mode() const;
 
+    // Seed the kvstore-backed local snapshot with a single state entry
+    // recording `mode`, before the partition (STM) starts, so a partition
+    // restored with a non-empty log comes up with partition_mode already set --
+    // before any leadership/config-propagation race could misclassify it. The
+    // seed survives persisted_stm start, whose initial cleanup only runs for an
+    // empty log (dirty_offset == {}), whereas a restored/bootstrapped partition
+    // has a non-empty log. `snapshot_offset` is the last-applied offset the
+    // seed represents (prev(log start)).
+    static ss::future<> seed_partition_mode(
+      storage::kvstore& kvstore,
+      ss::logger& logger,
+      const model::ntp& ntp,
+      model::offset snapshot_offset,
+      model::redpanda_storage_mode mode);
+
     // Registers a callback invoked on this shard whenever partition_mode may
     // have changed (on apply and on raft-snapshot restore), so the owner can
     // re-read partition_mode() and propagate it (e.g. into ntp_config).