From f9f1d027b91453114a017a9174b7e68a13ae0dff Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Mon, 29 Jun 2026 17:06:00 -0700 Subject: [PATCH 01/12] cluster: carry partition_mode in partition_properties_stm Persist a partition-local storage mode (model::redpanda_storage_mode) in the partition properties STM, alongside the existing writes_disabled property. This is the partition's own durable record of its storage mode, decoupled from the topic config; a later change exposes it on ntp_config as partition_mode (vs the topic-config-derived topic_mode) so the TS->CT migration can be modelled as reconciling partition_mode toward topic_mode. Adds update_partition_mode_cmd / operation_type::update_partition_mode, carries partition_mode on state_snapshot and raft_snapshot (serde version bump to v2, compat_version unchanged so older snapshots decode with partition_mode = unset), and the set_partition_mode / partition_mode / sync_partition_mode API. apply is refactored to dispatch on operation_type and copy-modify the current snapshot so each property update preserves the others. partition_mode updates are idempotent by value and ordered by log offset (no per-property revision). Also adds a change callback (set_partition_mode_change_callback), fired when partition_mode changes on apply or on raft-snapshot restore, so an owner can re-read partition_mode() and propagate it (the STM had no change notification; writes_disabled is polled). Tests cover: default unset, set/idempotent/change with cross-node consistency; partition_mode and writes_disabled are independent on the shared snapshot; raft snapshot carries partition_mode at the right offsets; recovery (local kvstore snapshot + replay, and raft-snapshot recovery after dropping a node's data); and the change callback fires on a real change but not on an idempotent no-op. Inert: no callers yet (the command is wired up and gated behind a feature in follow-up commits). Co-Authored-By: Claude Opus 4.8 --- src/v/cluster/partition_properties_stm.cc | 197 ++++++++++++++---- src/v/cluster/partition_properties_stm.h | 67 +++++- .../tests/partition_properties_stm_test.cc | 156 ++++++++++++++ 3 files changed, 380 insertions(+), 40 deletions(-) diff --git a/src/v/cluster/partition_properties_stm.cc b/src/v/cluster/partition_properties_stm.cc index 7e5392264a261..895781256d471 100644 --- a/src/v/cluster/partition_properties_stm.cc +++ b/src/v/cluster/partition_properties_stm.cc @@ -40,7 +40,8 @@ partition_properties_stm::partition_properties_stm( , _state_snapshots({state_snapshot{ .writes_disabled = writes_disabled::no, .update_offset = model::offset{}, - .writes_revision_id{}}}) {} + .writes_revision_id{}, + .partition_mode = model::redpanda_storage_mode::unset}}) {} ss::future partition_properties_stm::take_raft_snapshot(model::offset o) { @@ -88,7 +89,8 @@ partition_properties_stm::take_raft_snapshot(model::offset o) { co_return serde::to_iobuf( raft_snapshot{ .writes_disabled = it->writes_disabled, - .writes_revision_id = it->writes_revision_id}); + .writes_revision_id = it->writes_revision_id, + .partition_mode = it->partition_mode}); } ss::future @@ -157,43 +159,66 @@ void partition_properties_stm::apply_record( auto key = r.release_key(); auto value = r.release_value(); auto ot = serde::from_iobuf(std::move(key)); - if (ot != operation_type::update_writes_disabled) [[unlikely]] { + const auto update_offset = model::offset( + r.offset_delta() + batch_begin_offset); + const auto current_state = _state_snapshots.back(); + switch (ot) { + case operation_type::update_writes_disabled: { + auto update = serde::from_iobuf( + std::move(value)); vlog( - _log.warn, - "ignored unknown operation type with value: {}", - static_cast(ot)); + _log.trace, + "Applying update {} at offset: {}", + update, + update_offset); + bool is_legacy_command = update.writes_revision_id + == model::revision_id{}; + bool is_newer_revision = update.writes_revision_id + > current_state.writes_revision_id; + if (!is_newer_revision && !is_legacy_command) { + vlog( + _log.error, + "Ignoring out-of-order update: {}, current state: {}", + update, + current_state); + return; + } + bool differs = update.writes_disabled != current_state.writes_disabled + || update.writes_revision_id + != current_state.writes_revision_id; + if (differs) { + // copy-modify preserves the other properties (e.g. partition_mode) + auto next = current_state; + next.writes_disabled = update.writes_disabled; + next.writes_revision_id = update.writes_revision_id; + next.update_offset = update_offset; + _state_snapshots.push_back(next); + } return; } - auto update = serde::from_iobuf( - std::move(value)); - vlog( - _log.trace, - "Applying update {} at offset: {}", - update, - r.offset_delta() + batch_begin_offset); - const auto current_state = _state_snapshots.back(); - bool is_legacy_command = update.writes_revision_id == model::revision_id{}; - bool is_newer_revision = update.writes_revision_id - > current_state.writes_revision_id; - if (!is_newer_revision && !is_legacy_command) { + case operation_type::update_partition_mode: { + auto update = serde::from_iobuf( + std::move(value)); vlog( - _log.error, - "Ignoring out-of-order update: {}, current state: {}", + _log.trace, + "Applying partition_mode update {} at offset: {}", update, - _state_snapshots.back()); + update_offset); + // idempotent by value, ordered by log offset (no revision needed). + if (update.partition_mode != current_state.partition_mode) { + auto next = current_state; + next.partition_mode = update.partition_mode; + next.update_offset = update_offset; + _state_snapshots.push_back(next); + notify_partition_mode_change(); + } return; } - bool differs = update.writes_disabled != current_state.writes_disabled - || update.writes_revision_id - != current_state.writes_revision_id; - if (differs) { - _state_snapshots.push_back( - state_snapshot{ - .writes_disabled = update.writes_disabled, - .update_offset = model::offset( - r.offset_delta() + batch_begin_offset), - .writes_revision_id = update.writes_revision_id}); } + vlog( + _log.warn, + "ignored unknown operation type with value: {}", + static_cast(ot)); } ss::future<> @@ -211,7 +236,9 @@ partition_properties_stm::apply_raft_snapshot(const iobuf& buffer) { state_snapshot{ .writes_disabled = snapshot.writes_disabled, .update_offset = model::prev_offset(_raft->start_offset()), - .writes_revision_id = snapshot.writes_revision_id}); + .writes_revision_id = snapshot.writes_revision_id, + .partition_mode = snapshot.partition_mode}); + notify_partition_mode_change(); co_return; } @@ -301,6 +328,93 @@ model::record_batch partition_properties_stm::make_update_partitions_batch( return std::move(builder).build(); } +model::record_batch partition_properties_stm::make_update_partition_mode_batch( + update_partition_mode_cmd cmd) { + storage::record_batch_builder builder( + model::record_batch_type::partition_properties_update, model::offset{}); + builder.add_raw_kv( + serde::to_iobuf(operation_type::update_partition_mode), + serde::to_iobuf(std::move(cmd))); + return std::move(builder).build(); +} + +ss::future> +partition_properties_stm::replicate_partition_mode_update( + model::timeout_clock::duration timeout, update_partition_mode_cmd cmd) { + auto holder = _gate.hold(); + auto units = co_await _writes_mutex.get_units(); + if (!co_await sync(timeout)) { + co_return errc::not_leader; + } + + vassert( + !_state_snapshots.empty(), + "The invariant of state snapshot containing at least one element is " + "broken"); + auto current_state = _state_snapshots.back(); + if (cmd.partition_mode == current_state.partition_mode) [[unlikely]] { + // no-op + co_return current_state.update_offset; + } + + auto b = make_update_partition_mode_batch(cmd); + vlog(_log.debug, "replicating update partition mode command: {}", cmd); + raft::replicate_options r_opts( + raft::consistency_level::quorum_ack, + _insync_term, + std::chrono::milliseconds(timeout / 1ms)); + r_opts.set_force_flush(); + auto r = co_await _raft->replicate(std::move(b), r_opts); + + if (r.has_error()) { + vlog( + _log.warn, + "error replicating update partition mode command: {} - {}", + cmd, + r.error().message()); + co_await _raft->step_down("partition_properties_stm/replication_error"); + co_return r.error(); + } + + auto message_offset = r.value().last_offset; + if (!co_await wait_no_throw( + message_offset, model::timeout_clock::time_point::max())) { + co_await _raft->step_down("partition_properties_stm/replication_error"); + co_return errc::shutting_down; + } + co_return message_offset; +} + +ss::future> partition_properties_stm::set_partition_mode( + model::redpanda_storage_mode mode) { + vlog(_log.info, "setting partition mode to {}", mode); + return replicate_partition_mode_update( + _sync_timeout(), update_partition_mode_cmd{.partition_mode = mode}); +} + +model::redpanda_storage_mode partition_properties_stm::partition_mode() const { + vassert( + !_state_snapshots.empty(), + "The invariant of state snapshot containing at least one element is " + "broken"); + return _state_snapshots.back().partition_mode; +} + +void partition_properties_stm::notify_partition_mode_change() { + if (_partition_mode_change_cb) { + _partition_mode_change_cb(); + } +} + +ss::future> +partition_properties_stm::sync_partition_mode() { + auto holder = _gate.hold(); + if (!co_await sync(_sync_timeout())) { + co_return errc::not_leader; + } + co_return partition_mode(); +} + ss::future> partition_properties_stm::disable_writes(model::revision_id revision_id) { vlog(_log.info, "disabling partition writes"); @@ -355,9 +469,18 @@ fmt::iterator partition_properties_stm::raft_snapshot::format_to(fmt::iterator it) const { return fmt::format_to( it, - "{{writes_disabled: {}, writes_revision_id: {}}}", + "{{writes_disabled: {}, writes_revision_id: {}, partition_mode: {}}}", writes_disabled, - writes_revision_id); + writes_revision_id, + model::redpanda_storage_mode_to_string(partition_mode)); +} + +fmt::iterator partition_properties_stm::update_partition_mode_cmd::format_to( + fmt::iterator it) const { + return fmt::format_to( + it, + "{{partition_mode: {}}}", + model::redpanda_storage_mode_to_string(partition_mode)); } fmt::iterator partition_properties_stm::update_writes_disabled_cmd::format_to( @@ -373,10 +496,12 @@ fmt::iterator partition_properties_stm::state_snapshot::format_to(fmt::iterator it) const { return fmt::format_to( it, - "{{update_offset: {}, writes_disabled: {}, writes_revision_id: {}}}", + "{{update_offset: {}, writes_disabled: {}, writes_revision_id: {}, " + "partition_mode: {}}}", update_offset, writes_disabled, - writes_revision_id); + writes_revision_id, + model::redpanda_storage_mode_to_string(partition_mode)); } fmt::iterator diff --git a/src/v/cluster/partition_properties_stm.h b/src/v/cluster/partition_properties_stm.h index 32d4f635978cb..de8b5a4404ba7 100644 --- a/src/v/cluster/partition_properties_stm.h +++ b/src/v/cluster/partition_properties_stm.h @@ -15,9 +15,12 @@ #include "cluster/state_machine_registry.h" #include "container/chunked_vector.h" #include "model/fundamental.h" +#include "model/metadata.h" #include "raft/persisted_stm.h" #include "serde/envelope.h" +#include + namespace cluster { struct partition_properties_stm_accessor; @@ -50,6 +53,26 @@ class partition_properties_stm // Returns a current value of the writes disabled property. writes_disabled are_writes_disabled() const; + // Updates the persistent, partition-local storage mode; returns the offset + // of the update message. This is the partition's own durable record of its + // storage mode, decoupled from the topic config (see partition_mode()). + ss::future> + set_partition_mode(model::redpanda_storage_mode); + // Waits for the state to be up to date and returns the partition mode; only + // intended to be called on the current leader. + ss::future> sync_partition_mode(); + // Returns the current partition mode. `unset` means "not yet bootstrapped"; + // callers (ntp_config) fall back to the topic-config-derived mode. + model::redpanda_storage_mode partition_mode() const; + + // 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). + void + set_partition_mode_change_callback(ss::noncopyable_function cb) { + _partition_mode_change_cb = std::move(cb); + } + raft::stm_initial_recovery_policy get_initial_recovery_policy() const final { return raft::stm_initial_recovery_policy::skip_to_end; @@ -80,19 +103,43 @@ class partition_properties_stm const update_writes_disabled_cmd&) = default; }; + struct update_partition_mode_cmd + : serde::envelope< + update_partition_mode_cmd, + serde::version<0>, + serde::compat_version<0>> { + model::redpanda_storage_mode partition_mode + = model::redpanda_storage_mode::unset; + + auto serde_fields() { return std::tie(partition_mode); } + fmt::iterator format_to(fmt::iterator it) const; + friend bool operator==( + const update_partition_mode_cmd&, + const update_partition_mode_cmd&) = default; + }; + struct state_snapshot : serde::envelope< state_snapshot, - serde::version<1>, + serde::version<2>, serde::compat_version<0>> { // todo: in a major release we can change it to use a local_snapshot + // update_offset writes_disabled writes_disabled; model::offset update_offset; model::revision_id writes_revision_id; + // serde v2; absent in older snapshots -> defaults to `unset`, which the + // partition_mode() accessor / ntp_config treats as "fall back to the + // topic-config-derived mode" (bootstrap fallback). + model::redpanda_storage_mode partition_mode + = model::redpanda_storage_mode::unset; auto serde_fields() { - return std::tie(writes_disabled, update_offset, writes_revision_id); + return std::tie( + writes_disabled, + update_offset, + writes_revision_id, + partition_mode); } fmt::iterator format_to(fmt::iterator it) const; friend bool @@ -101,12 +148,15 @@ class partition_properties_stm struct raft_snapshot : serde:: - envelope, serde::compat_version<0>> { + envelope, serde::compat_version<0>> { writes_disabled writes_disabled = writes_disabled::no; model::revision_id writes_revision_id; + model::redpanda_storage_mode partition_mode + = model::redpanda_storage_mode::unset; auto serde_fields() { - return std::tie(writes_disabled, writes_revision_id); + return std::tie( + writes_disabled, writes_revision_id, partition_mode); } fmt::iterator format_to(fmt::iterator it) const; @@ -134,6 +184,7 @@ class partition_properties_stm enum class operation_type { update_writes_disabled = 0, + update_partition_mode = 1, }; ss::future<> do_apply(const model::record_batch&) final; @@ -142,10 +193,15 @@ class partition_properties_stm static model::record_batch make_update_partitions_batch(update_writes_disabled_cmd); + static model::record_batch + make_update_partition_mode_batch(update_partition_mode_cmd); ss::future> replicate_properties_update( model::timeout_clock::duration timeout, update_writes_disabled_cmd command); + ss::future> replicate_partition_mode_update( + model::timeout_clock::duration timeout, + update_partition_mode_cmd command); config::binding _sync_timeout; @@ -158,6 +214,9 @@ class partition_properties_stm // For linearizing access to writes_disabled state on the leader ssx::mutex _writes_mutex{"c/partition_properties_stm::writes_mutex"}; + + void notify_partition_mode_change(); + ss::noncopyable_function _partition_mode_change_cb; }; class partition_properties_stm_factory : public state_machine_factory { diff --git a/src/v/cluster/tests/partition_properties_stm_test.cc b/src/v/cluster/tests/partition_properties_stm_test.cc index 40ecdab618656..4caef6b73c842 100644 --- a/src/v/cluster/tests/partition_properties_stm_test.cc +++ b/src/v/cluster/tests/partition_properties_stm_test.cc @@ -16,6 +16,7 @@ #include "config/mock_property.h" #include "container/chunked_circular_buffer.h" #include "model/fundamental.h" +#include "model/metadata.h" #include "model/record.h" #include "model/tests/random_batch.h" #include "raft/replicate.h" @@ -115,6 +116,32 @@ struct partition_properties_stm_fixture : raft::raft_fixture { ASSERT_EQ_CORO(r.value(), disabled); } + ss::future<> set_partition_mode( + model::redpanda_storage_mode mode, bool expect_success = true) { + auto res = co_await retry_with_leader( + raft::default_timeout(), + 2s, + [mode](raft::raft_node_instance& leader_node) { + return get_stm(leader_node)->set_partition_mode(mode); + }); + ASSERT_EQ_CORO(res.has_value(), expect_success); + } + + ss::future<> assert_partition_mode(model::redpanda_storage_mode mode) { + auto r = co_await get_leader_stm()->sync_partition_mode(); + ASSERT_TRUE_CORO(r.has_value()); + ASSERT_EQ_CORO(r.value(), mode); + } + + ss::future<> + check_partition_mode_consistent(model::redpanda_storage_mode mode) { + RPTEST_REQUIRE_EVENTUALLY_CORO(5s, [this, mode] { + return std::ranges::all_of( + node_stms | std::views::values, + [mode](auto stm) { return stm->partition_mode() == mode; }); + }); + } + ss::future> replicate_random_batches() { return retry_with_leader( raft::default_timeout(), @@ -417,4 +444,133 @@ TEST_F_CORO( co_await check_state_consistent(should_be_disabled); } +TEST_F_CORO(partition_properties_stm_fixture, test_partition_mode_basic) { + co_await initialize_state_machines(); + co_await wait_for_leader(10s); + // A fresh partition has never had partition_mode written: it reads `unset` + // (the "not bootstrapped -> fall back to topic_mode" sentinel). + co_await assert_partition_mode(model::redpanda_storage_mode::unset); + + co_await set_partition_mode(model::redpanda_storage_mode::tiered); + co_await assert_partition_mode(model::redpanda_storage_mode::tiered); + // idempotent + co_await set_partition_mode(model::redpanda_storage_mode::tiered); + co_await assert_partition_mode(model::redpanda_storage_mode::tiered); + // the migration cutover transition: tiered -> cloud + co_await set_partition_mode(model::redpanda_storage_mode::cloud); + co_await assert_partition_mode(model::redpanda_storage_mode::cloud); + co_await check_partition_mode_consistent( + model::redpanda_storage_mode::cloud); +} + +// writes_disabled and partition_mode share one state_snapshot; updating one +// must preserve the other (the apply copy-modify). +TEST_F_CORO( + partition_properties_stm_fixture, test_partition_mode_independent_of_writes) { + co_await initialize_state_machines(); + + co_await disable_writes(model::revision_id{1}); + co_await set_partition_mode(model::redpanda_storage_mode::tiered); + co_await assert_writes(stm_t::writes_disabled::yes); + co_await assert_partition_mode(model::redpanda_storage_mode::tiered); + + // a partition_mode update leaves writes_disabled untouched + co_await set_partition_mode(model::redpanda_storage_mode::cloud); + co_await assert_writes(stm_t::writes_disabled::yes); + co_await assert_partition_mode(model::redpanda_storage_mode::cloud); + + // and a writes_disabled update leaves partition_mode untouched + co_await enable_writes(model::revision_id{2}); + co_await assert_writes(stm_t::writes_disabled::no); + co_await assert_partition_mode(model::redpanda_storage_mode::cloud); +} + +TEST_F_CORO(partition_properties_stm_fixture, test_partition_mode_snapshot) { + co_await initialize_state_machines(); + + auto before_set = co_await replicate_random_batches(); + co_await set_partition_mode(model::redpanda_storage_mode::tiered); + [[maybe_unused]] auto _ignored = co_await replicate_random_batches(); + + auto leader_id = get_leader(); + EXPECT_TRUE(leader_id.has_value()); + auto& leader_node = node(leader_id.value()); + + // snapshot taken before the set command -> unset + auto o = co_await leader_node.random_batch_base_offset(before_set.value()); + auto snap_before = partition_properties_stm_accessor::snap_from_iobuf( + co_await get_leader_stm()->take_raft_snapshot(o)); + ASSERT_EQ_CORO( + snap_before.partition_mode, model::redpanda_storage_mode::unset); + + // snapshot at the set command offset -> tiered (and writes_disabled intact) + auto snap_at = partition_properties_stm_accessor::snap_from_iobuf( + co_await get_leader_stm()->take_raft_snapshot( + model::next_offset(before_set.value()))); + ASSERT_EQ_CORO( + snap_at.partition_mode, model::redpanda_storage_mode::tiered); + ASSERT_EQ_CORO(snap_at.writes_disabled, stm_t::writes_disabled::no); +} + +TEST_F_CORO(partition_properties_stm_fixture, test_partition_mode_recovery) { + co_await initialize_state_machines(); + co_await set_partition_mode(model::redpanda_storage_mode::tiered); + [[maybe_unused]] auto _ignored = co_await replicate_random_batches(); + co_await set_partition_mode(model::redpanda_storage_mode::cloud); + co_await assert_partition_mode(model::redpanda_storage_mode::cloud); + auto last_applied = get_leader_stm()->last_applied_offset(); + + // restart preserving data: recover from local (kvstore) snapshot + replay + co_await restart_nodes(); + co_await wait_for_leader(10s); + co_await assert_partition_mode(model::redpanda_storage_mode::cloud); + co_await wait_for_committed_offset(last_applied, 10s); + co_await check_partition_mode_consistent( + model::redpanda_storage_mode::cloud); + + // take a raft snapshot on every node, then restart with one node's data + // dropped: that node recovers partition_mode from the raft snapshot. + for (auto& [_, node] : nodes()) { + auto base_offset = co_await node->random_batch_base_offset( + node->raft()->committed_offset(), model::offset(1)); + auto snapshot_offset = model::prev_offset(base_offset); + auto result = co_await node->raft()->stm_manager()->take_snapshot( + snapshot_offset); + co_await node->raft()->write_snapshot( + raft::write_snapshot_cfg(snapshot_offset, std::move(result.data))); + } + co_await restart_nodes({model::node_id(1)}); + co_await wait_for_leader(10s); + co_await assert_partition_mode(model::redpanda_storage_mode::cloud); + co_await wait_for_committed_offset(last_applied, 10s); + co_await check_partition_mode_consistent( + model::redpanda_storage_mode::cloud); +} + +// The change callback (used by partition to push partition_mode into +// ntp_config) must fire when partition_mode changes, on each replica that +// applies the change. +TEST_F_CORO( + partition_properties_stm_fixture, test_partition_mode_change_callback) { + co_await initialize_state_machines(); + co_await wait_for_leader(10s); + + auto fired = ss::make_lw_shared(0); + for (auto& [_, stm] : node_stms) { + stm->set_partition_mode_change_callback([fired] { ++(*fired); }); + } + + co_await set_partition_mode(model::redpanda_storage_mode::tiered); + RPTEST_REQUIRE_EVENTUALLY_CORO(5s, [fired] { return *fired >= 1; }); + co_await check_partition_mode_consistent( + model::redpanda_storage_mode::tiered); + + // an idempotent (no-op) update does not change state, so it does not fire + auto before = *fired; + co_await set_partition_mode(model::redpanda_storage_mode::tiered); + co_await check_partition_mode_consistent( + model::redpanda_storage_mode::tiered); + ASSERT_EQ_CORO(*fired, before); +} + } // namespace cluster From e69f9f2a7d472f95d5eb9a27bfbafe0395c13777 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Mon, 29 Jun 2026 18:33:04 -0700 Subject: [PATCH 02/12] features: add the tiered_to_cloud_migration feature Declare feature::tiered_to_cloud_migration (bit 21, v26_2_1). It gates all behaviors of the TS->CT migration capability across the PR stack: the partition_mode write introduced in this base branch, and the migration trigger/cutover in the later PRs. The feature is `always`, so it auto-activates once the cluster reaches its release version; whether a migration may actually be triggered is separately gated behind the enable_ts_tsv2_migration config (added later in the stack). NOTE: kept at v26.2.1 (the current MAX release version) so it auto-enables today. Before merge this should target the next release version (26.3), which requires adding that release version separately. Co-Authored-By: Claude Opus 4.8 --- src/v/features/feature_table.cc | 2 ++ src/v/features/feature_table.h | 7 +++++++ 2 files changed, 9 insertions(+) diff --git a/src/v/features/feature_table.cc b/src/v/features/feature_table.cc index 2877725743dd2..b405f3d86198b 100644 --- a/src/v/features/feature_table.cc +++ b/src/v/features/feature_table.cc @@ -111,6 +111,8 @@ std::string_view to_string_view(feature f) { return "remote_labels"; case feature::partition_properties_stm: return "partition_properties_stm"; + case feature::tiered_to_cloud_migration: + return "tiered_to_cloud_migration"; case feature::shadow_indexing_split_topic_property_update: return "shadow_indexing_split_topic_property_update"; case feature::datalake_iceberg: diff --git a/src/v/features/feature_table.h b/src/v/features/feature_table.h index 1128d0f4aeeae..6ae115d144ebb 100644 --- a/src/v/features/feature_table.h +++ b/src/v/features/feature_table.h @@ -66,6 +66,7 @@ enum class feature : std::uint64_t { membership_change_controller_cmds = 1ULL << 22U, controller_snapshots = 1ULL << 23U, cloud_storage_manifest_format_v2 = 1ULL << 24U, + tiered_to_cloud_migration = 1ULL << 25U, force_partition_reconfiguration = 1ULL << 26U, delete_records = 1ULL << 29U, raft_coordinated_recovery = 1ULL << 31U, @@ -431,6 +432,12 @@ inline constexpr std::array feature_schema{ feature::partition_properties_stm, feature_spec::available_policy::always, feature_spec::prepare_policy::always}, + feature_spec{ + release_version::v26_2_1, + "tiered_to_cloud_migration", + feature::tiered_to_cloud_migration, + feature_spec::available_policy::always, + feature_spec::prepare_policy::always}, feature_spec{ release_version::v24_3_1, "shadow_indexing_split_topic_property_update", From 36c855d5cdbc489793e5bf57f8eb4895dff770ba Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Mon, 29 Jun 2026 17:46:43 -0700 Subject: [PATCH 03/12] storage: split ntp_config storage mode into topic_mode and partition_mode Introduce two storage-mode accessors on ntp_config: - topic_mode() the mode the topic config requests (the existing _overrides->storage_mode; the "desired" mode). - partition_mode() the partition's own durable mode, fed from partition_properties via set_partition_mode(); the "actual" mode. Falls back to topic_mode() when unset. The serving/behavior predicates (is_archival_enabled, is_remote_fetch_enabled, is_tiered_storage, cloud_topic_enabled, is_tiered_cloud) now key on partition_mode() instead of reading _overrides->storage_mode directly. The legacy shadow_indexing fallback (used when the mode is unset) is preserved unchanged, since archival-only/fetch-only SI modes have no redpanda_storage_mode equivalent. TS->CT migration will be modelled as partition_mode lagging topic_mode (still `tiered` while the topic config says `cloud`) until cutover, with serving keyed on the actual mode. Co-Authored-By: Claude Opus 4.8 --- src/v/storage/ntp_config.h | 77 ++++++++++++++++++++++++-------------- 1 file changed, 49 insertions(+), 28 deletions(-) diff --git a/src/v/storage/ntp_config.h b/src/v/storage/ntp_config.h index a2328fadcdb06..4b34a556f350d 100644 --- a/src/v/storage/ntp_config.h +++ b/src/v/storage/ntp_config.h @@ -247,38 +247,58 @@ class ntp_config { : topic_recovery_enabled::no; } + // The storage mode the topic config requests (the "desired" mode). `unset` + // for a legacy topic that uses shadow_indexing_mode instead of + // storage_mode. + model::redpanda_storage_mode topic_mode() const { + return _overrides ? _overrides->storage_mode + : model::redpanda_storage_mode::unset; + } + + // The partition's own durable storage mode (the "actual" mode), recorded in + // partition_properties and pushed in via set_partition_mode(). Falls back + // to topic_mode() when unset (not yet bootstrapped). Serving/behavior + // accessors key on partition_mode rather than topic_mode. + // On storage mode switch, partition_mode will lag topic_mode until the + // migration completes. + model::redpanda_storage_mode partition_mode() const { + return _partition_mode != model::redpanda_storage_mode::unset + ? _partition_mode + : topic_mode(); + } + + void set_partition_mode(model::redpanda_storage_mode m) { + _partition_mode = m; + } + bool is_archival_enabled() const { - if (_overrides == nullptr) { - return false; - } + const auto mode = partition_mode(); // Explicit tiered - if (_overrides->storage_mode == model::redpanda_storage_mode::tiered) { + if (mode == model::redpanda_storage_mode::tiered) { return true; } // Explicit local or cloud - if (_overrides->storage_mode != model::redpanda_storage_mode::unset) { + if (mode != model::redpanda_storage_mode::unset) { return false; } // Unset, fall back to legacy shadow_indexing - return _overrides->shadow_indexing_mode + return _overrides && _overrides->shadow_indexing_mode && model::is_archival_enabled( _overrides->shadow_indexing_mode.value()); } bool is_remote_fetch_enabled() const { - if (_overrides == nullptr) { - return false; - } + const auto mode = partition_mode(); // Explicit tiered - if (_overrides->storage_mode == model::redpanda_storage_mode::tiered) { + if (mode == model::redpanda_storage_mode::tiered) { return true; } // Explicit local or cloud - if (_overrides->storage_mode != model::redpanda_storage_mode::unset) { + if (mode != model::redpanda_storage_mode::unset) { return false; } // Unset, fall back to legacy shadow_indexing - return _overrides->shadow_indexing_mode + return _overrides && _overrides->shadow_indexing_mode && model::is_fetch_enabled( _overrides->shadow_indexing_mode.value()); } @@ -302,23 +322,22 @@ class ntp_config { * both reads and writes to S3, and is not a read replica. */ bool is_tiered_storage() const { - if (_overrides == nullptr) { - return false; - } - if (_overrides->read_replica.value_or(false)) { + if (is_read_replica_mode_enabled()) { return false; } + const auto mode = partition_mode(); // Explicit tiered - if (_overrides->storage_mode == model::redpanda_storage_mode::tiered) { + if (mode == model::redpanda_storage_mode::tiered) { return true; } // Explicit local or cloud - if (_overrides->storage_mode != model::redpanda_storage_mode::unset) { + if (mode != model::redpanda_storage_mode::unset) { return false; } // Unset, fall back to legacy shadow_indexing - return _overrides->shadow_indexing_mode - == model::shadow_indexing_mode::full; + return _overrides + && _overrides->shadow_indexing_mode + == model::shadow_indexing_mode::full; } bool remote_delete() const { @@ -439,17 +458,13 @@ class ntp_config { } bool cloud_topic_enabled() const { - return _overrides - && (_overrides->storage_mode - == model::redpanda_storage_mode::cloud - || _overrides->storage_mode - == model::redpanda_storage_mode::tiered_cloud); + const auto mode = partition_mode(); + return mode == model::redpanda_storage_mode::cloud + || mode == model::redpanda_storage_mode::tiered_cloud; } bool is_tiered_cloud() const { - return _overrides - && _overrides->storage_mode - == model::redpanda_storage_mode::tiered_cloud; + return partition_mode() == model::redpanda_storage_mode::tiered_cloud; } std::optional min_cleanable_dirty_ratio() const { @@ -511,6 +526,12 @@ class ntp_config { /// _topic_revision in case of recovered topics or read replicas. model::initial_revision_id _remote_rev{0}; + /// The partition's own durable storage mode, fed from partition_properties + /// (see partition_mode()). `unset` until bootstrapped, when the mode + /// accessors fall back to the topic-config-derived topic_mode(). + model::redpanda_storage_mode _partition_mode + = model::redpanda_storage_mode::unset; + public: // in storage/types.cc fmt::iterator format_to(fmt::iterator it) const; From 542dda39f8abf553d85e3d4e79be2479973e8787 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Mon, 29 Jun 2026 18:04:27 -0700 Subject: [PATCH 04/12] cluster/partition: feed partition_mode from partition_properties into ntp_config On partition start, read partition_properties_stm::partition_mode() and push it into the log's ntp_config (storage::log gains a set_partition_mode method, mirroring set_overrides), and register the STM's change callback so the value is re-pushed whenever the STM applies a partition_mode update or restores a raft snapshot. ntp_config::partition_mode() therefore tracks the partition's durable mode, falling back to topic_mode() while unset. No feature guard here: reading replicated STM state and pushing it into the local ntp_config is always safe. Writing partition_mode (which requires the tiered_to_cloud_migration feature) is added in the bootstrap commit; until then the value is always unset. Co-Authored-By: Claude Opus 4.8 --- src/v/cluster/partition.cc | 19 +++++++++++++++++++ src/v/cluster/partition.h | 5 +++++ src/v/storage/disk_log_impl.cc | 4 ++++ src/v/storage/disk_log_impl.h | 1 + src/v/storage/log.h | 4 ++++ 5 files changed, 33 insertions(+) diff --git a/src/v/cluster/partition.cc b/src/v/cluster/partition.cc index 5dd95cc675f99..853d62060267c 100644 --- a/src/v/cluster/partition.cc +++ b/src/v/cluster/partition.cc @@ -473,6 +473,17 @@ ss::future<> partition::start( _partition_properties_stm = _raft->stm_manager()->get(); + // 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); @@ -1759,6 +1770,14 @@ 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::set_writes_disabled( partition_properties_stm::writes_disabled disable, model::timeout_clock::time_point deadline, diff --git a/src/v/cluster/partition.h b/src/v/cluster/partition.h index 5a3950c8915f2..f01e955887d28 100644 --- a/src/v/cluster/partition.h +++ b/src/v/cluster/partition.h @@ -425,6 +425,11 @@ class partition : public ss::enable_lw_shared_from_this { // 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(); + consensus_ptr _raft; // never null ss::shared_ptr _log_eviction_stm; ss::shared_ptr _rm_stm; diff --git a/src/v/storage/disk_log_impl.cc b/src/v/storage/disk_log_impl.cc index 78f2cc21a9d49..c1827b13b9291 100644 --- a/src/v/storage/disk_log_impl.cc +++ b/src/v/storage/disk_log_impl.cc @@ -3799,6 +3799,10 @@ void disk_log_impl::set_overrides(ntp_config::default_overrides o) { mutable_config().set_overrides(o); } +void disk_log_impl::set_partition_mode(model::redpanda_storage_mode m) { + mutable_config().set_partition_mode(m); +} + /** * We express compaction backlog as the size of data that has to be read to * perform full compaction. diff --git a/src/v/storage/disk_log_impl.h b/src/v/storage/disk_log_impl.h index 85093de19110a..84f3651e517f1 100644 --- a/src/v/storage/disk_log_impl.h +++ b/src/v/storage/disk_log_impl.h @@ -168,6 +168,7 @@ class disk_log_impl final : public log { size_t size_bytes() const override { return _probe->partition_size(); } uint64_t size_bytes_after_offset(model::offset o) const override; void set_overrides(ntp_config::default_overrides) final; + void set_partition_mode(model::redpanda_storage_mode) final; bool notify_compaction_update() final; int64_t compaction_backlog() final; diff --git a/src/v/storage/log.h b/src/v/storage/log.h index dc0b86a8b9428..083775f650585 100644 --- a/src/v/storage/log.h +++ b/src/v/storage/log.h @@ -223,6 +223,10 @@ class log { /// topic/partition-level overrides virtual void set_overrides(ntp_config::default_overrides) = 0; + /// Mutates the partition_mode stored in the log's ntp_config: the + /// partition's own durable storage mode, fed from partition_properties. + virtual void set_partition_mode(model::redpanda_storage_mode) {} + /// Notifies the log about a possible change to the log compaction config. /// Returns true if the log compaction changed. virtual bool notify_compaction_update() = 0; From 176249e4116581f741a12bb44487bc53f3412492 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Sun, 12 Jul 2026 16:46:27 -0700 Subject: [PATCH 05/12] cluster/partition: bootstrap partition_mode on leadership On becoming leader with the tiered_to_cloud_migration feature active, record the partition's storage mode into partition_properties if not already set (maybe_bootstrap_partition_mode, one-time and idempotent). It keys on the archival manifest (archival_metadata_stm::holds_archived_data() -- a node that became leader after an in-place ts->ct switch is still tiered while topic_mode() already reads cloud), else falls back to topic_mode(); a legacy shadow_indexing topic (topic_mode unset) is left unset. Wired to partition_manager's leadership notification. Co-Authored-By: Claude Opus 4.8 --- .../cluster/archival/archival_metadata_stm.cc | 5 +++ .../cluster/archival/archival_metadata_stm.h | 5 +++ src/v/cluster/partition.cc | 41 +++++++++++++++++++ src/v/cluster/partition.h | 9 ++++ src/v/cluster/partition_manager.cc | 6 +++ 5 files changed, 66 insertions(+) diff --git a/src/v/cluster/archival/archival_metadata_stm.cc b/src/v/cluster/archival/archival_metadata_stm.cc index 401dbcc1a5afc..b076839637c89 100644 --- a/src/v/cluster/archival/archival_metadata_stm.cc +++ b/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()) { diff --git a/src/v/cluster/archival/archival_metadata_stm.h b/src/v/cluster/archival/archival_metadata_stm.h index faa555b236a81..dd9f42afc1930 100644 --- a/src/v/cluster/archival/archival_metadata_stm.h +++ b/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( diff --git a/src/v/cluster/partition.cc b/src/v/cluster/partition.cc index 853d62060267c..90c8a622fff54 100644 --- a/src/v/cluster/partition.cc +++ b/src/v/cluster/partition.cc @@ -1778,6 +1778,47 @@ 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`. + // - topic_mode(): the partition's configured storage mode. A legacy + // shadow_indexing topic has topic_mode() == unset and is left unset. + const bool holds_ts_data = _archival_meta_stm + && _archival_meta_stm->holds_archived_data(); + const auto target = holds_ts_data + ? model::redpanda_storage_mode::tiered + : 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> partition::set_writes_disabled( partition_properties_stm::writes_disabled disable, model::timeout_clock::time_point deadline, diff --git a/src/v/cluster/partition.h b/src/v/cluster/partition.h index f01e955887d28..f0d626652fe78 100644 --- a/src/v/cluster/partition.h +++ b/src/v/cluster/partition.h @@ -164,6 +164,15 @@ class partition : public ss::enable_lw_shared_from_this { 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(); + bool has_followers() const; void block_new_leadership() const; void unblock_new_leadership() const; diff --git a/src/v/cluster/partition_manager.cc b/src/v/cluster/partition_manager.cc index c19f563c187cf..6b49c5678d30d 100644 --- a/src/v/cluster/partition_manager.cc +++ b/src/v/cluster/partition_manager.cc @@ -30,6 +30,7 @@ #include "raft/consensus_utils.h" #include "raft/fundamental.h" #include "ssx/async-clear.h" +#include "ssx/future-util.h" #include #include @@ -73,6 +74,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( From 421f318c7b5c3783b5c647ba7c914956e6683a9c Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Sun, 12 Jul 2026 16:46:44 -0700 Subject: [PATCH 06/12] cluster/partition: bootstrap partition_mode on migration-feature activation The leadership bootstrap is gated on the feature being active, so a partition whose leader was elected while tiered_to_cloud_migration was inactive keeps partition_mode == unset. On feature activation, sweep every current leader and re-run the (idempotent) maybe_bootstrap_partition_mode (bootstrap_partition_mode_on_migration_feature, awaited from partition_manager::start); partitions that gain leadership after activation are covered by the leadership notification. Co-Authored-By: Claude Opus 4.8 --- src/v/cluster/partition_manager.cc | 27 +++++++++++++++++++++++++++ src/v/cluster/partition_manager.h | 5 +++++ 2 files changed, 32 insertions(+) diff --git a/src/v/cluster/partition_manager.cc b/src/v/cluster/partition_manager.cc index 6b49c5678d30d..aa94da2ca33c6 100644 --- a/src/v/cluster/partition_manager.cc +++ b/src/v/cluster/partition_manager.cc @@ -106,9 +106,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 partition_manager::manage( storage::ntp_config ntp_cfg, raft::group_id group, diff --git a/src/v/cluster/partition_manager.h b/src/v/cluster/partition_manager.h index edd4691df0e1f..7fbcf6c997f38 100644 --- a/src/v/cluster/partition_manager.h +++ b/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_manager; From 3079698b5f801be7c45bf0974d81642aa7192225 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Thu, 11 Jun 2026 16:06:25 -0700 Subject: [PATCH 07/12] cloud_topics/stm: don't pin truncation on an idle ctp_stm In order to dynamically switch a partition from TS to CT/TSv2, we need to be able to enable the ctp_stm on the fly. The simplest way to do that is to simply attach an empty ctp_stm to every partition. Unfortunately, with LRO unset, ctp_stm previously returned offset::min() for get_max_collectible_offset() preventing log trimming. This commit modifies that behavior to return offset::max() if _max_applied_epoch is empty making the presence of a ctp_stm harmless until advance_epoch() is called. Co-Authored-By: Claude Opus 4.8 --- src/v/cloud_topics/level_zero/stm/ctp_stm_state.cc | 11 ++++++++++- .../level_zero/stm/tests/ctp_stm_state_test.cc | 11 ++++++++++- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/src/v/cloud_topics/level_zero/stm/ctp_stm_state.cc b/src/v/cloud_topics/level_zero/stm/ctp_stm_state.cc index 93ba62caf65df..b8d0708bd36cd 100644 --- a/src/v/cloud_topics/level_zero/stm/ctp_stm_state.cc +++ b/src/v/cloud_topics/level_zero/stm/ctp_stm_state.cc @@ -179,7 +179,16 @@ model::offset ctp_stm_state::get_max_collectible_offset() const noexcept { if (_last_reconciled_log_offset.has_value()) { return _last_reconciled_log_offset.value(); } - // Truncation is impossible without LRO + // An STM that has applied no CT data must not constrain truncation: it may + // be an inert passenger on a partition that is not (yet) a cloud topic + // (e.g. a tiered partition pre-installed with ctp_stm so a migration needs + // no runtime STM install). Without this a passenger would pin the local log + // at offset::min() forever. + if (!_max_applied_epoch.has_value()) { + return model::offset::max(); + } + // CT data has been applied but not yet reconciled to L1: protect it. + // Truncation is impossible without an LRO. return model::offset::min(); } diff --git a/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_state_test.cc b/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_state_test.cc index acd039e44ade7..57f182c7c6d87 100644 --- a/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_state_test.cc +++ b/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_state_test.cc @@ -30,7 +30,9 @@ TEST(ctp_stm_state_test, initial_state) { EXPECT_FALSE(state.get_max_seen_epoch(model::term_id(1)).has_value()); EXPECT_FALSE(state.get_last_reconciled_offset().has_value()); EXPECT_FALSE(state.get_last_reconciled_log_offset().has_value()); - EXPECT_EQ(state.get_max_collectible_offset(), model::offset::min()); + // An STM with no applied CT data is an inert passenger and must not + // constrain truncation. + EXPECT_EQ(state.get_max_collectible_offset(), model::offset::max()); } TEST(ctp_stm_state_test, advance_max_seen_epoch) { @@ -107,8 +109,15 @@ TEST(ctp_stm_state_test, advance_last_reconciled_offset) { TEST(ctp_stm_state_test, get_max_collectible_offset) { ct::ctp_stm_state state; + // No CT data applied yet: inert passenger, do not constrain truncation. + EXPECT_EQ(state.get_max_collectible_offset(), model::offset::max()); + + // CT data applied (a placeholder advanced the epoch) but not yet reconciled + // to L1: protect it by pinning collection at min(). + state.advance_epoch(ct::cluster_epoch(1), model::offset(10)); EXPECT_EQ(state.get_max_collectible_offset(), model::offset::min()); + // Once reconciled, collect up to the reconciled log offset. model::offset log_offset(500); state.advance_last_reconciled_offset(kafka::offset(300), log_offset); EXPECT_EQ(state.get_max_collectible_offset(), log_offset); From 2b446a0745b802779cf1edfc02f66468b6c9555a Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Thu, 11 Jun 2026 16:06:25 -0700 Subject: [PATCH 08/12] cloud_topics/stm: pre-install an idle ctp_stm on tiered-storage partitions is_applicable_for: pre-install the (idle) ctp_stm on tiered-storage partitions when cloud topics are available, so a tiered->cloud migration needs no runtime STM install -- the STM is already present and inert until the partition becomes a cloud topic. STM membership is fixed at construction, so it has to be there beforehand. Co-Authored-By: Claude Opus 4.8 --- src/v/cloud_topics/level_zero/stm/BUILD | 1 + src/v/cloud_topics/level_zero/stm/ctp_stm.cc | 10 ++++++ .../level_zero/stm/ctp_stm_factory.cc | 16 +++++++-- .../level_zero/stm/tests/ctp_stm_test.cc | 35 +++++++++++++++++++ 4 files changed, 60 insertions(+), 2 deletions(-) diff --git a/src/v/cloud_topics/level_zero/stm/BUILD b/src/v/cloud_topics/level_zero/stm/BUILD index 7b19efc9cd26f..f2b99619394d8 100644 --- a/src/v/cloud_topics/level_zero/stm/BUILD +++ b/src/v/cloud_topics/level_zero/stm/BUILD @@ -182,6 +182,7 @@ redpanda_cc_library( implementation_deps = [ ":ctp_stm", "//src/v/cloud_topics:logger", + "//src/v/config", ], visibility = ["//visibility:public"], deps = ["//src/v/cluster:state_machine_registry"], diff --git a/src/v/cloud_topics/level_zero/stm/ctp_stm.cc b/src/v/cloud_topics/level_zero/stm/ctp_stm.cc index fea23c7be9477..ff5a328de448c 100644 --- a/src/v/cloud_topics/level_zero/stm/ctp_stm.cc +++ b/src/v/cloud_topics/level_zero/stm/ctp_stm.cc @@ -427,6 +427,16 @@ ctp_stm::take_local_snapshot(ssx::semaphore_units) { } ss::future<> ctp_stm::apply_raft_snapshot(const iobuf& buf) { + // An empty snapshot is applied during recovery fast-forward: the raft + // snapshot created when bootstrapping a pre-existing partition carries no + // per-STM data, so state_machine_manager hands each STM an empty buffer to + // advance over. There is no ctp_stm state to restore then -- e.g. a + // partition recovered mid tiered->cloud migration has no L1 state (its data + // is still in tiered storage). Keep the default (empty) state and advance, + // mirroring log_eviction_stm's empty-snapshot handling. + if (buf.empty()) { + co_return; + } auto snap = serde::from_iobuf(buf.copy()); _state = std::move(snap.state); _epoch_checker = snap.checker; diff --git a/src/v/cloud_topics/level_zero/stm/ctp_stm_factory.cc b/src/v/cloud_topics/level_zero/stm/ctp_stm_factory.cc index 6d29d0ceb0175..1714309acba33 100644 --- a/src/v/cloud_topics/level_zero/stm/ctp_stm_factory.cc +++ b/src/v/cloud_topics/level_zero/stm/ctp_stm_factory.cc @@ -12,13 +12,25 @@ #include "cloud_topics/level_zero/stm/ctp_stm.h" #include "cloud_topics/logger.h" +#include "config/configuration.h" namespace cloud_topics::l0 { bool ctp_stm_factory::is_applicable_for( const storage::ntp_config& ntp_cfg) const { - return ntp_cfg.cloud_topic_enabled() - && !ntp_cfg.is_read_replica_mode_enabled(); + if (ntp_cfg.is_read_replica_mode_enabled()) { + return false; + } + if (ntp_cfg.cloud_topic_enabled()) { + return true; + } + // Pre-install an (idle) ctp_stm on tiered-storage partitions when cloud + // topics are available, so a tiered->cloud/tiered_cloud migration needs no + // runtime STM install -- the STM is already present and inert + // (get_max_collectible_offset() returns max() until CT data is applied) + // until the partition becomes a cloud topic. + return config::shard_local_cfg().cloud_storage_enabled() + && ntp_cfg.is_archival_enabled(); } void ctp_stm_factory::create( diff --git a/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_test.cc b/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_test.cc index f5b79b8e3ca27..32db1e0679518 100644 --- a/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_test.cc +++ b/src/v/cloud_topics/level_zero/stm/tests/ctp_stm_test.cc @@ -48,6 +48,10 @@ struct ctp_stm_accessor { snapshot.header, std::move(snapshot.data)); } + auto apply_raft_snapshot(ctp_stm& stm, const iobuf& buf) { + return stm.apply_raft_snapshot(buf); + } + bool epoch_cv_has_waiters(ctp_stm& stm) { return stm._epoch_updated_cv.has_waiters(); } @@ -564,6 +568,37 @@ TEST_F_CORO(ctp_stm_fixture, test_snapshot) { } } +// The recovery fast-forward hands each STM an empty raft snapshot to advance +// over: a bootstrapped pre-existing partition carries no per-STM data, and a +// partition recovered mid tiered->cloud migration has no L1 ctp_stm state (its +// data is still in tiered storage). apply_raft_snapshot must treat an empty +// buffer as a no-op -- not try to deserialize it (which threw before the guard) +// and not clobber existing state with a default. +TEST_F_CORO(ctp_stm_fixture, apply_empty_raft_snapshot_is_noop) { + co_await start(); + co_await wait_for_leader(raft::default_timeout()); + auto& leader = node(*get_leader()); + + // Establish some applied state so we can prove the empty snapshot doesn't + // reset it. + auto b1 = make_record_batch(ct::cluster_epoch{2}, model::offset{0}, 0); + auto res = co_await replicate_record_batch(leader, std::move(b1)); + ASSERT_TRUE_CORO(res.has_value()); + auto max_epoch_before = api(leader).get_max_epoch(); + ASSERT_TRUE_CORO(max_epoch_before.has_value()); + ASSERT_EQ_CORO(max_epoch_before.value(), ct::cluster_epoch{2}); + + auto stm = get_stm<0>(leader); + ct::ctp_stm_accessor a; + // Must not throw on an empty buffer. + co_await a.apply_raft_snapshot(*stm, iobuf{}); + + // State preserved, not reset to default. + auto max_epoch_after = api(leader).get_max_epoch(); + ASSERT_TRUE_CORO(max_epoch_after.has_value()); + ASSERT_EQ_CORO(max_epoch_after.value(), max_epoch_before.value()); +} + TEST_F_CORO(ctp_stm_fixture, test_fence_epoch_concurrent_new_epoch) { // This test verifies the optimization in fence_epoch() where multiple // concurrent requests for a new epoch only require one write lock. From ad0e71f238a6a7fe99418629169f70a6622404c1 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Fri, 5 Jun 2026 14:55:00 -0700 Subject: [PATCH 09/12] archival: create the archival STM on cloud-topic partitions During migration, we need the newly configured TSv2 or CT partition to continue to behave as an archival stm. Thus, we need the archival stm to continue to exist after the config switch. The simplest way to do this is to allow it to exist generally on CT/TSv2 topics. Drop the cloud_topic_enabled() == false guard from archival_metadata_stm_factory::is_applicable_for, so every cloud-topic partition carries an archival STM. Co-Authored-By: Claude Opus 4.8 --- src/v/cluster/archival/archival_metadata_stm.cc | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/v/cluster/archival/archival_metadata_stm.cc b/src/v/cluster/archival/archival_metadata_stm.cc index b076839637c89..cce025d6bf555 100644 --- a/src/v/cluster/archival/archival_metadata_stm.cc +++ b/src/v/cluster/archival/archival_metadata_stm.cc @@ -1734,10 +1734,14 @@ archival_metadata_stm_factory::archival_metadata_stm_factory( bool archival_metadata_stm_factory::is_applicable_for( const storage::ntp_config& ntp_cfg) const { + // The archival STM is created on cloud-topic partitions too (no + // cloud_topic_enabled() == false guard). For a partition migrated from + // tiered storage it is reconstructed from its snapshot and its manifest + // stays available to gate local-log truncation. For a partition that was + // always a cloud topic the manifest is empty, so it is an inert passenger. return _cloud_storage_enabled && _cloud_storage_api.local_is_initialized() && ntp_cfg.ntp().tp.topic != model::kafka_consumer_offsets_topic - && ntp_cfg.ntp().ns == model::kafka_namespace - && ntp_cfg.cloud_topic_enabled() == false; + && ntp_cfg.ntp().ns == model::kafka_namespace; } void archival_metadata_stm_factory::create( From 12e3fe07c905c2414a88218e672b594d38f12b87 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Thu, 11 Jun 2026 21:04:48 -0700 Subject: [PATCH 10/12] archival: clamp local-log trim while the manifest holds archived data A cloud-mode partition whose storage mode has been flipped reports is_archival_enabled() == false, which would make max_removable_local_log_offset collect_all and stop constraining local-log truncation -- evicting tiered-storage data not yet uploaded. Keep constraining via cloud_recoverable_offset() while the partition still holds archived data (a non-empty live manifest or a spillover archive), introduced here as holds_archived_data(); once the manifest is emptied, trimming resumes. Co-Authored-By: Claude Opus 4.8 --- .../cluster/archival/archival_metadata_stm.cc | 12 +++++++ .../cluster/archival/archival_metadata_stm.h | 1 + .../tests/archival_metadata_stm_test.cc | 36 +++++++++++++++++++ 3 files changed, 49 insertions(+) diff --git a/src/v/cluster/archival/archival_metadata_stm.cc b/src/v/cluster/archival/archival_metadata_stm.cc index cce025d6bf555..b62eb6cb65003 100644 --- a/src/v/cluster/archival/archival_metadata_stm.cc +++ b/src/v/cluster/archival/archival_metadata_stm.cc @@ -1373,6 +1373,18 @@ model::offset archival_metadata_stm::max_removable_local_log_offset() { collect_all = false; } + // The manifest, not the storage mode, governs local-log truncation. While + // the partition still holds archived data (a non-empty live manifest or a + // spillover archive) it is still served from tiered storage and may hold + // local data not yet uploaded, so its local log must not be trimmed past + // cloud_recoverable_offset(). is_archival_enabled() can report false while + // archived data is still present -- e.g. a partition mid tiered->cloud + // migration whose effective storage mode is cloud -- which would otherwise + // collect_all and stop constraining truncation, evicting the not-yet- + // uploaded data. Gate collect_all on holds_archived_data() so it takes + // effect only once the manifest is cleared. + collect_all = collect_all && !holds_archived_data(); + if (collect_all || is_read_replica || (uploads_paused && gaps_allowed)) { // The archival is disabled but the state machine still exists so we // shouldn't stop eviction from happening. diff --git a/src/v/cluster/archival/archival_metadata_stm.h b/src/v/cluster/archival/archival_metadata_stm.h index dd9f42afc1930..320dc07bf1fc7 100644 --- a/src/v/cluster/archival/archival_metadata_stm.h +++ b/src/v/cluster/archival/archival_metadata_stm.h @@ -257,6 +257,7 @@ class archival_metadata_stm final : public raft::persisted_stm<> { model::offset get_last_offset() const; model::offset get_archive_start_offset() const; model::offset get_archive_clean_offset() const; + kafka::offset get_start_kafka_offset() const; // Return list of all segments that has to be diff --git a/src/v/cluster/archival/tests/archival_metadata_stm_test.cc b/src/v/cluster/archival/tests/archival_metadata_stm_test.cc index 2c758c5ce29ed..6ac60c30c64a6 100644 --- a/src/v/cluster/archival/tests/archival_metadata_stm_test.cc +++ b/src/v/cluster/archival/tests/archival_metadata_stm_test.cc @@ -256,6 +256,42 @@ FIXTURE_TEST( archival_stm->manifest().begin()->committed_offset, model::offset(99)); } +// With archival disabled (the flipped, mid-migration storage mode) but a +// non-empty manifest, max_removable_local_log_offset keeps constraining +// local-log truncation rather than collecting all -- protecting tiered-storage +// data that has not yet been uploaded. +FIXTURE_TEST(test_migration_local_trim_clamp, archival_metadata_stm_fixture) { + wait_for_confirmed_leader(); + + // The fixture's partition has archival disabled. With an empty manifest, + // nothing constrains local-log truncation. + BOOST_REQUIRE_EQUAL( + archival_stm->max_removable_local_log_offset(), model::offset::max()); + + // Add a segment so the manifest is non-empty. + std::vector m; + m.push_back( + segment_meta{ + .base_offset = model::offset(0), + .committed_offset = model::offset(99), + .archiver_term = model::term_id(1), + .segment_term = model::term_id(1)}); + archival_stm + ->add_segments( + m, + std::nullopt, + model::producer_id{}, + ss::lowres_clock::now() + 10s, + never_abort, + cluster::segment_validated::yes) + .get(); + BOOST_REQUIRE_EQUAL(archival_stm->manifest().size(), 1); + + // Now truncation is constrained (no longer offset::max()). + BOOST_REQUIRE_NE( + archival_stm->max_removable_local_log_offset(), model::offset::max()); +} + FIXTURE_TEST(test_archival_stm_segment_replace, archival_metadata_stm_fixture) { wait_for_confirmed_leader(); std::vector m1; From 7d4ae19ffd2d267acda59834702a64fb5a5c27d8 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Thu, 11 Jun 2026 21:05:00 -0700 Subject: [PATCH 11/12] cluster: install the eviction STM on cloud-topic partitions Drop the cloud_topic_enabled() guard from log_eviction_stm_factory so the eviction STM is installed on cloud-topic partitions too. A partition migrating from tiered storage is still served from its local log + archival manifest and must keep supporting prefix truncation (DeleteRecords) while migrating; the trigger flips storage_mode to cloud without reconstructing the partition, and a still-migrating partition is reconstructed cloud-mode on restart/recovery, so the STM must already be present. On a native cloud topic it is an inert passenger. Subsequent integration tests will provide coverage. Co-Authored-By: Claude Opus 4.8 --- src/v/cluster/log_eviction_stm.cc | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/v/cluster/log_eviction_stm.cc b/src/v/cluster/log_eviction_stm.cc index 931da1e6bd758..7f3d8852cd0ce 100644 --- a/src/v/cluster/log_eviction_stm.cc +++ b/src/v/cluster/log_eviction_stm.cc @@ -384,9 +384,15 @@ log_eviction_stm_factory::log_eviction_stm_factory(storage::kvstore& kvstore) bool log_eviction_stm_factory::is_applicable_for( const storage::ntp_config& cfg) const { - if (cfg.cloud_topic_enabled()) { - return false; - } + // (No cloud_topic_enabled() == false guard.) Install the eviction STM on + // cloud-topic partitions too. STM membership is fixed at construction and a + // tiered->cloud migration does not reconstruct the partition, so a + // partition migrating from tiered storage -- still served from its local + // log + archival manifest -- must already carry the eviction STM to keep + // supporting prefix truncation (DeleteRecords) and retention while + // migrating. On a native cloud topic it is an inert passenger: retention + // and DeleteRecords go through the L1 path, so no local eviction requests + // are generated. return !storage::deletion_exempt(cfg.ntp()); } From 1c764c974b5b88a59d072d01ac09b8e3731064c0 Mon Sep 17 00:00:00 2001 From: Samuel Just Date: Thu, 23 Jul 2026 11:47:41 -0700 Subject: [PATCH 12/12] cloud_topics/stm: add explicit-log-offset advance_reconciled_offset overload Split advance_reconciled_offset so the reconciled log offset can be supplied explicitly instead of always deriving it from the raft offset translator. The existing kafka-offset-only entry point now delegates, deriving the log offset as before; both share the min_allowed_local_threshold handling. This lets a caller seed the reconciliation baseline of a TS-migrated partition directly to the migration boundary -- passing the precise raft offset of the last uploaded tiered-storage segment -- so the already reconciled region can be trimmed without re-deriving the offset. ctp_stm carries no migration state of its own; this is expressed purely as "reconciliation starts at offset N". Co-Authored-By: Claude Opus 4.8 --- src/v/cloud_topics/level_zero/stm/ctp_stm_api.cc | 13 ++++++++++++- src/v/cloud_topics/level_zero/stm/ctp_stm_api.h | 14 ++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/v/cloud_topics/level_zero/stm/ctp_stm_api.cc b/src/v/cloud_topics/level_zero/stm/ctp_stm_api.cc index d40c37f6909a1..a21f9290e2682 100644 --- a/src/v/cloud_topics/level_zero/stm/ctp_stm_api.cc +++ b/src/v/cloud_topics/level_zero/stm/ctp_stm_api.cc @@ -122,6 +122,18 @@ ctp_stm_api::advance_reconciled_offset( kafka::offset lro, model::timeout_clock::time_point deadline, ss::abort_source& as, + std::optional min_allowed_local_threshold) { + auto lrlo = _stm->_raft->log()->to_log_offset(kafka::offset_cast(lro)); + return advance_reconciled_offset( + lro, lrlo, deadline, as, min_allowed_local_threshold); +} + +ss::future> +ctp_stm_api::advance_reconciled_offset( + kafka::offset lro, + model::offset lrlo, + model::timeout_clock::time_point deadline, + ss::abort_source& as, std::optional min_allowed_local_threshold) { // The floor is monotonic: drop targets the state already covers. if ( @@ -138,7 +150,6 @@ ctp_stm_api::advance_reconciled_offset( model::record_batch_type::ctp_stm_command, model::offset(0)); if (advances_lro) { - auto lrlo = _stm->_raft->log()->to_log_offset(kafka::offset_cast(lro)); vlog( _log.debug, "Replicating ctp_stm_cmd::advance_reconciled_offset{{lro:{} " diff --git a/src/v/cloud_topics/level_zero/stm/ctp_stm_api.h b/src/v/cloud_topics/level_zero/stm/ctp_stm_api.h index 4b31ec5369a0e..2762e78cfcc9d 100644 --- a/src/v/cloud_topics/level_zero/stm/ctp_stm_api.h +++ b/src/v/cloud_topics/level_zero/stm/ctp_stm_api.h @@ -85,6 +85,20 @@ class ctp_stm_api { ss::abort_source& as, std::optional min_allowed_local_threshold = std::nullopt); + /// As above, but with the reconciled log offset supplied explicitly rather + /// than derived from the raft offset translator. Used to seed the + /// reconciliation baseline of a TS-migrated partition directly to the + /// migration boundary (kafka offset + the raft offset of the last uploaded + /// TS segment) so the already-uploaded TS region of the local raft log can + /// be trimmed without waiting for the first CT reconciliation cycle. + ss::future> + advance_reconciled_offset( + kafka::offset last_reconciled_offset, + model::offset last_reconciled_log_offset, + model::timeout_clock::time_point deadline, + ss::abort_source& as, + std::optional min_allowed_local_threshold = std::nullopt); + ss::future> set_start_offset( kafka::offset new_start_offset,