RFC ts-ct-migration: migration mirror + archiver lifecycle [PR 5]#30893
Draft
sjust-redpanda wants to merge 31 commits into
Draft
RFC ts-ct-migration: migration mirror + archiver lifecycle [PR 5]#30893sjust-redpanda wants to merge 31 commits into
sjust-redpanda wants to merge 31 commits into
Conversation
This was referenced Jun 23, 2026
sjust-redpanda
force-pushed
the
sjust/ts-ct-migration-pr5
branch
3 times, most recently
from
47abecb to
3fdf567
Compare
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 <noreply@anthropic.com>
…tions 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 <noreply@anthropic.com>
…verload 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. 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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
remote_segment_index.{h,cc} contained two independent things sharing a TU: the
read/write `offset_index` (used by readers) and `remote_segment_index_builder`, a
storage::batch_consumer that pulls in raft (raft::offset_translator_batch_types).
offset_index itself has none of that coupling -- only model, bytes, utils/delta_for
and serde.
Move `offset_index` into cloud_storage/offset_index.{h,cc} behind a new
//src/v/cloud_storage:offset_index cc_library.
remote_segment_index.{h,cc} keeps segment_record_stats + the builder, now
including offset_index.h; the cloud_storage monolith depends on :offset_index.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add an optional descriptor that lets an L1 object/extent point at a segment that physically lives in the tiered-storage bucket instead of an L1-native object. Metadata/encoding only; no read, write, or delete path consumes it yet. - imported_ts_info (object_id.h): the per-segment descriptor -- ts_path, delta_offset/delta_offset_end (log<->kafka offset translation), segment_term, and last_kafka_offset. Stored split by ownership: ts_path on the object row (imported_ts_object_location), delta/term on the extent row (imported_ts_segment_info); absent means a native L1 object. - Thread it through the metastore read/interface types on both backends: an optional imported_ts_info on object_response / extent_object_info (recomposed from the storage split on read), plus rpc_types and the replicated/domain passthrough. - Encode it: serde-versioned on both backends (object_entry/extent and the LSM rows), the LSM debug-serde path, and the ImportedTsObjectLocation / ImportedTsSegmentInfo protobufs. Old-version decode defaults to absent. Testing: serde round-trip unit tests for the new fields, including old-version decode. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add a durable, per-partition boolean migrating to each backend's partition record, set true while a partition is mid tiered->cloud migration. - state: add the field to partition_state (serde v0->v1) and the LSM metadata_row_value (v1->v2), defaulting to false so snapshots predating the field decode as native cloud topics; carry it through the MetadataValue debug proto. - op: add set_migrating(partition, bool) across the metastore interface, the RPC service/client, the leader router (+probe), both domain managers, and both backends. It is idempotent (setting the current value is a no-op), applies in either direction, and creates the partition record when set on an absent partition; metadata-row rewrites (add_objects, set_start_offset) preserve the flag. - read: surface the flag on get_offsets. Tests (both backends): create-on-absent, set-then-clear, idempotent set, and preservation across add_objects; serde back-compat for both records. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Let the metastore register tiered-storage segments as imported L1 extents
by reference, reusing the existing add_objects extent-addition path rather
than a dedicated command.
- add_objects now handles imported objects:
- skips the pre-registration check for imported objects (they reference an
existing tiered-storage segment; nothing is written to L1)
- copies imported_ts_location / imported_ts_delta onto the object and
extent rows
- sets start_offset from the first extent for an empty migrating partition
(a non-zero start -- the TS->CT mount), rather than requiring offset 0
- the object builder's add_imported(ntp_metadata) registers one imported
segment as a finished, single-extent object: with no L1 write to reserve it
creates the object id and records it as finished without pre-registration,
to be committed by a subsequent add_objects. meta.imported must be set; the
read-replica metastore rejects it.
Tests (both backends): a migrating partition adopts a non-zero start at the
first imported extent; successive batches forward-append; a non-contiguous
batch is rejected.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
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 <noreply@anthropic.com>
Declare feature::tiered_to_cloud_migration (bit 18, explicit_only, 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. explicit_only means it does not auto-activate and carries no mixed-version compatibility commitment until an operator opts in. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…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 wil lbe 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 <noreply@anthropic.com>
… 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 <noreply@anthropic.com>
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>
…artition A partition served from tiered storage whose local log does not cover its cloud data (e.g. mid tiered->cloud migration after cluster recovery wiped the local log) reports high watermark 0, hiding the recovered prefix from ListOffsets and consumers. Report max(local, next_cloud_offset()) when cloud data is available; for a healthy partition the cloud watermark never exceeds the local one, so this is a no-op. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Assert that flipping a topic's storage mode to cloud (topic_mode) while the durable partition_mode lags at tiered leaves the partition served as tiered storage with no observable change: replicated_partition offsets / size estimates / timequery are byte-for-byte unchanged, make_partition_proxy still routes to the replicated_partition path, and prefix truncation still works and advances the start. Regresses the timequery corner (must consult the imported prefix, not the local-log start). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
sjust-redpanda
force-pushed
the
sjust/ts-ct-migration-pr5
branch
from
3fdf567 to
da70144
Compare
… abstraction
Introduce an object_handle/object_index indirection for the L1 read path so a
reader can seek and stream from an object without knowing its on-disk format.
This is the seam a later commit uses to serve imported tiered-storage segments;
here it is wired for native L1 only and is behavior-preserving.
- object_handle.{h,cc}: object_index (seek_to_offset/seek_to_timestamp ->
seek_result) and object_handle (index() + open_reader) interfaces, plus the
native implementations l1_footer_index and l1_native_object_handle. The
handle reads byte ranges via a fetch callback over io::read_object, so file_io
and fake_io share one implementation.
- open_object.{h,cc}: a free open_object() factory that reads the object footer
via io::fetch_native_footer and returns an l1_native_object_handle
(footer-indexed) whose fetch callback reads byte ranges back through
io::read_object.
- io::read_object_as_iobuf, whose sole use is reading the footer, is renamed to
fetch_native_footer; file_io overrides it to record footer bytes on its probe
(the footer_read_bytes metric).
- level_one_reader obtains an object_handle via open_object and reads through
its index/open_reader instead of reading the footer directly; group_id is
threaded through.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add chunk_data_source: an ss::data_source_impl that fetches a byte range one fixed-size chunk at a time, lazily, behind a single contiguous input_stream. Only the chunks a read actually consumes are downloaded -- when the reader stops at its limit it stops pulling, and the tail is never fetched. Chunks are fixed byte ranges (not aligned to any structure in the data); the source concatenates one chunk's tail with the next chunk's head, so a record spanning a boundary is read seamlessly, and EOF is signalled only at the true end (never mistaken for a chunk boundary). It is a templated header-only library, generic over the fetch callable and its error type, so it depends on nothing L1-specific; make_chunk_data_source deduces the fetch type and wraps it in an ss::data_source. No caller yet uses it; the imported tiered-storage read path adopts it in the following commit. Tested: chunk_data_source_test covers the source directly (concatenation byte-identical to a whole download, mid-chunk head skip, early-termination fetches only touched chunks, single/exact-multiple sizes, EOF at total length, fetch-error propagation). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add the building blocks for serving an imported tiered-storage segment, not yet
wired to any io backend (inert on their own):
- tiered_storage_object_reader (ts_reader.{h,cc}): an object_reader over a raw
Kafka-format segment stream. Maintains a running offset delta, emits only data
batches, drops aborted-transaction ranges (from the .tx manifest) so the
region is committed-only, and stamps the segment term as each batch's leader
epoch.
- ts_segment_index (ts_object.{h,cc}): object_index over a segment's
offset_index -- translates a target offset/timestamp into a byte position +
delta via find_kaf_offset/find_timestamp.
- ts_object_handle (ts_object.{h,cc}): object_handle whose byte transport is
injected as a fetch_range_fn, so file_io (download + cache) and fake_io
(in-memory) can share the seek/reader wiring. open_reader serves the segment
as a lazily-fetched sequence of fixed-size chunks through the
ts_chunk_data_source (only the chunks a read touches are downloaded);
chunk_size == 0 disables chunking and fetches the whole suffix in one range.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Extend the open_object factory to dispatch an imported extent (one carrying imported_ts_info) to the tiered-storage read path, so a single L1 reader serves native L1 objects and imported segments interchangeably. The segment's index (.index) and aborted-transaction ranges (.tx manifest) are fetched through two new io methods, fetch_ts_index and fetch_ts_tx. They return raw index bytes and parsed tx ranges rather than cloud_storage domain objects, so the widely-depended-on abstract_io interface stays free of a cloud_storage dependency. open_object assembles them into a ts_segment_index and an aborted set, gates the .tx fetch on the import-time tx_state, and returns a ts_object_handle whose byte transport reads chunks back through io::read_object (routed to the segment's ts_path, with cache/skip handling). - file_io: read_object routes an imported extent to its ts_path; fetch_ts_index / fetch_ts_tx download the .index and .tx sidecars (via a download_raw_iobuf helper) and record their byte counts on the probe. - fake_io: an injected segment (put_ts_segment, optionally with a serialized offset_index) is served through the same path with an in-memory fetch; without an index the ts_segment_index is empty (a full-segment scan), matching file_io's missing-.index fallback. The imported chunk size is taken from config (cloud_storage_cache_chunk_size / cloud_storage_disable_chunk_reads); tests set it small via scoped_config to exercise multi-chunk reads. Testing: imported open_object/reader unit tests (offset translation, the index-backed seek, read-committed strip, leader-epoch stamping, control-batch drop) against fake_io, plus the imported reader fixture. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
When an imported L1 object is removed, delete its backing tiered-storage objects (the segment, its .tx range manifest, and its .index -- the same set the archiver would have deleted) so nothing is orphaned. - delete_objects (abstract_io/file_io/fake_io) takes object_location instead of object_id, so it routes by the imported location: present -> delete from the tiered-storage bucket; absent -> delete from the L1 bucket as before. - Both garbage collectors (metastore and lsm) build the removal list from each object's imported_ts_location. Native L1 deletion is unchanged. Testing: an imported-segment GC test (import via add_objects on a migrating partition, then trim + GC) asserts the backing TS segment is removed, routed by ts_path; fake_io grows has_ts_segment to observe it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Factor the collector's delete-the-backing-objects-then-drop-the-metastore-rows step into a remove_objects() method on both the simple and LSM garbage collectors; remove_unreferenced delegates to it. Sync the STM before building the removal update. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Derive the single authoritative "mid tiered->cloud migration" signal from the two storage modes: the topic requests a cloud mode (topic_mode) while the partition's durable partition_mode still lags at tiered. Behaviors that must key on the migration phase -- rather than the eventual serving mode or the manifest -- read this. Replaces the archival STM migration flag. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
A partition mid tiered->cloud migration must keep an archiver to upload, GC, and mirror its tiered data. It is tiered-mode -- partition_mode lags at tiered until cutover -- so should_construct_archiver's normal is_archival_enabled() path builds it; no cloud-topic special case is needed. On recovery partition_mode can restore from the log tail after start()/leadership, too late for the construction check there, so re-evaluate maybe_start_archiver() on two triggers: leadership acquisition (partition_manager) and partition_mode changing (the partition_properties change callback). A recovered migrating partition then resumes its migration mirror regardless of when partition_mode is applied. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The migration mirror runs as an archiver housekeeping job (leader-only, co-located with the manifest), but the L1 metastore client lives in cloud_topics, which depends on cluster -- not the reverse. So the archiver cannot call the metastore directly without inverting the module dependency. Introduce archival::migration_metastore, an abstract sink (model types only) the archiver writes through to mirror tiered-storage segments into L1: forward-append imported segments, prune (detach) the head, query the partition's L1 offsets, and mark the offline phase complete. cloud_topics::migration_metastore_sink implements it by translating to the L1 metastore client (replicated_metastore). A later change injects an instance into the archiver and adds the mirror fiber that drives it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Thread the migration metastore sink from the cloud-topics subsystem down to partitions so the archiver's mirror job can reach it. partition_manager holds a migration_metastore* and propagates it to current and future partitions; each partition exposes it via migration_metastore() for the archiver to read lazily. The cloud-topics app constructs a sharded migration_metastore_sink over the L1 metastore client and registers it with the partition manager on startup, deregistering it on stop so no partition retains a dangling pointer. The pointer is null until the cloud-topics subsystem registers it; the mirror is simply inactive until then. No behavior change yet -- the mirror job that uses the sink is added next. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
While a partition is migrating (ntp_config::is_migrating() -- the topic requests cloud but partition_mode still lags at tiered), the archiver dark-copies its tiered-storage manifest into the cloud-topics L1 metastore via an injected sink, trimming the front and appending the tail so a converged partition can cut over. The mirror and migration head-prune run only while is_migrating(). Because the mirror imports by trim-front + append-tail and cannot reflect an in-place segment rewrite, two manifest-mutating jobs are suspended while is_migrating(): compacted reupload and adjacent segment merging. Both key on is_migrating() rather than the storage mode: partition_mode lags at tiered during migration, so cloud_topic_enabled()/is_archival_enabled() report the tiered values and would let these jobs run. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
PR2 pre-installs an idle ctp_stm on tiered-storage partitions, so the housekeeper now manages partitions that are not (yet) cloud topics -- a plain tiered partition, or one still migrating tiered->cloud (partition_mode lags at tiered until cutover). Running CT housekeeping against such a partition forces epoch/placeholder maintenance that seeds a meaningless reconciled offset and pins local-log GC, freezing retention. Gate the loop on cloud_topic_enabled() so it runs only on cloud topics, skipping until cutover. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
sjust-redpanda
force-pushed
the
sjust/ts-ct-migration-pr5
branch
from
da70144 to
f0a678a
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
🚧 NOT YET READY FOR REVIEW
Summary
Add the background machinery that, while a partition is migrating, keeps its archiver running and continuously mirrors the archival manifest into the L1 metastore as imported extents (by reference, no copy) — the cutover-last "dark mirror". Inert until the migration trigger is enabled (a later PR).
Stacked on PR1 + PR3 + PR4. Base is
dev, so the diff is cumulative (it includes PR1–PR4); commits new in this PR: construct-archiver, the metastore sink + its injection, the mirror pass, skip-CT-housekeeping-while-migrating, local-retention-while-migrating, and the case-C flag.What's in this PR
should_construct_archiverbuilds an archiver for a cloud-mode partition with a non-empty archival manifest, so its tiered-storage data keeps being uploaded/GC'd/mirrored; the two in-place segment-rewrite jobs (adjacent merging, compacted reupload) are suspended while cloud-mode (the mirror only trims-front and appends-tail).migration_metastore(interface incluster, model types only) implemented by acloud_topicssink and wiredpartition_manager → partition → archiver, so the archiver writes L1 without a module dependency inversion.full_log_start_kafka_offset), and forward-append the rest of the log (live manifest + spillover sub-manifests) as imported extents. Per-segment resolution is a testable free functionmake_imported_segmentthat also resolves the.tx-manifest presence (as native TS would) and guards the pre-offset-translationdelta_offset == min()sentinel.Testing
migration_segment_testwalks the per-segment resolution matrix (v1/v2/v3 × compacted × metadata_size_hint, the sentinel, the kafka-empty skip).housekeeper_testcovers skipping CT housekeeping while migrating.🤖 Generated with Claude Code
Design Doc
PR 0 · PR 1 · PR 2 · PR 3 · PR 4 · PR 5 · PR 6 · PR 7 · PR 8