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 71e1ea732f3ef..120fe03f292ee 100644 --- a/src/v/cloud_topics/level_zero/stm/ctp_stm.cc +++ b/src/v/cloud_topics/level_zero/stm/ctp_stm.cc @@ -424,6 +424,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_api.cc b/src/v/cloud_topics/level_zero/stm/ctp_stm_api.cc index bad61748cb4d1..9f9d65b2ecf58 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 @@ -121,13 +121,21 @@ ss::future> ctp_stm_api::advance_reconciled_offset( kafka::offset lro, model::timeout_clock::time_point deadline, + ss::abort_source& as) { + auto lrlo = _stm->_raft->log()->to_log_offset(kafka::offset_cast(lro)); + return advance_reconciled_offset(lro, lrlo, deadline, as); +} + +ss::future> +ctp_stm_api::advance_reconciled_offset( + kafka::offset lro, + model::offset lrlo, + model::timeout_clock::time_point deadline, ss::abort_source& as) { if (lro <= get_last_reconciled_offset()) { co_return std::monostate{}; } - auto lrlo = _stm->_raft->log()->to_log_offset(kafka::offset_cast(lro)); - vlog( _log.debug, "Replicating ctp_stm_cmd::advance_reconciled_offset{{lro:{} lrlo:{}}}", 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 9aed35f8b1c0f..5b9b03af512be 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 @@ -76,6 +76,19 @@ class ctp_stm_api { model::timeout_clock::time_point deadline, ss::abort_source& as); + /// Advance the reconciled offset with an explicit log offset, rather than + /// deriving it 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); + ss::future> set_start_offset( kafka::offset new_start_offset, 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/ctp_stm_state.cc b/src/v/cloud_topics/level_zero/stm/ctp_stm_state.cc index 4907e834ffd1b..d38678ae2788a 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 @@ -151,7 +151,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 8dd5ad44b807b..156e44475fc8b 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); 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 e252a1a30cf7f..50576d7e3a5cf 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(); } @@ -521,6 +525,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. diff --git a/src/v/cluster/archival/archival_metadata_stm.cc b/src/v/cluster/archival/archival_metadata_stm.cc index 401dbcc1a5afc..1e4c6694550e6 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. @@ -1690,6 +1702,11 @@ model::offset archival_metadata_stm::get_archive_start_offset() const { return _manifest->get_archive_start_offset(); } +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_archive_clean_offset() const { return _manifest->get_archive_clean_offset(); } @@ -1729,10 +1746,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( diff --git a/src/v/cluster/archival/archival_metadata_stm.h b/src/v/cluster/archival/archival_metadata_stm.h index faa555b236a81..e94169cc6009b 100644 --- a/src/v/cluster/archival/archival_metadata_stm.h +++ b/src/v/cluster/archival/archival_metadata_stm.h @@ -252,6 +252,10 @@ 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; + + /// True if the partition still has data in tiered storage -- either in the + /// live STM manifest or offloaded to the spillover archive. + bool holds_archived_data() 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; 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()); }