You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "hubcio (via GitHub)" <gi...@apache.org> on 2026/08/01 17:07:09 UTC
Re: [PR] feat(metadata): state transfer for snapshot and client table [iggy]
hubcio commented on code in PR #3765:
URL: https://github.com/apache/iggy/pull/3765#discussion_r3696087104
##########
core/metadata/src/stm/mod.rs:
##########
@@ -434,6 +440,10 @@ macro_rules! collect_handlers {
$crate::stm::StateHandler::apply(payload, self, *ts)
},
)*
+ [<$state Command>]::RestoreSnapshot(snapshot) => {
+ *self = Self::inner_from_snapshot(snapshot.clone());
Review Comment:
`inner_from_snapshot` mints a fresh `Arc::new(StatsRegistry::default())` per call, and this arm runs once per left-right buffer (`absorb_first` and `absorb_second` both dispatch it), so after an install the two buffers hold different registries. boot is only safe because `from_snapshot` builds one inner and `new_from_empty` clones it. worse, `inner_from_snapshot` never repopulates the registry's `partitions` map, so every `Arc<PartitionStats>` the data plane registered at bootstrap/reconcile is orphaned - `partition_get` returns none and partition stats read (0,0,0,0) permanently on any node that state-transferred, while stream/topic aggregates alternate between the two arc sets by publish parity. the comment three lines above the `Arc::new` in `stream.rs` states exactly the invariant this breaks.
fix: pass the existing shared registry into the rebuild (this arm has `self` to source it) and prune it to the snapshot's key set while keeping survivors' partition entries - slab keys get reused, so a stale entry would hand a new stream a dead one's counters. note the only test pinning exact counts is `cfg(not(feature = "vsr"))`, so CI can't see this.
##########
core/shard/src/lib.rs:
##########
@@ -2161,7 +2311,34 @@ where
&& consensus.namespace() == header.namespace
{
match consensus.handle_commit(&header) {
- CommitOutcome::Advanced => planes.0.commit_journal().await,
+ CommitOutcome::Advanced => {
+ // Mid-transfer the pre-install STM must not walk: the
+ // snapshot being installed already contains those ops.
+ // `commit_max` still advanced inside `handle_commit`, so
+ // the post-install repair targets the right frontier.
+ if !consensus.is_transferring() {
Review Comment:
this gate is right, but the same walk is reachable ungated from `on_start_view` (line 2225) and `on_do_view_change` (line 2128) - the `AwaitingTarget` early-return only covers that one stage, so a StartView arriving during `Fetching` walks the pre-install STM. two consequences: the walk can advance `commit_min` past the incoming `snapshot_seq`, flipping `snapshot_ahead` to false so the install silently degrades to table-only (no STM restore, no persist, no pairing) while still logging success; and if the advance lands inside the install's superblock await, `set_commit_floor`'s anti-rewind assert panics the shard. same one-line gate belongs on both paths.
##########
core/shard/src/lib.rs:
##########
@@ -2749,6 +2964,790 @@ where
.await;
}
+ /// Start metadata tail journal-repair from `peer` when the commit walk
+ /// gap-stopped below the known frontier. Shared by `StartView` adoption
+ /// and the post-install step of a state transfer.
+ #[allow(clippy::future_not_send)]
+ async fn maybe_request_metadata_repair<P>(&self, consensus: &VsrConsensus<B, P>, peer: u8)
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ if consensus.is_normal()
+ && !consensus.is_transferring()
+ && consensus.commit_min() < consensus.commit_max()
+ && self.metadata_repair.borrow().is_none()
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ let to_op = consensus.commit_max();
+ let from_op = consensus.commit_min() + 1;
+ *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
+ nonce,
+ to_op,
+ peer,
+ idle_ticks: 0,
+ });
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ "metadata behind the group frontier; requesting repair"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
+
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn send_request_state_transfer<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ target: u8,
+ nonce: u128,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let msg =
+ Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
+ .transmute_header(|_, h: &mut RequestStateTransferHeader| {
+ h.command = Command2::RequestStateTransfer;
+ h.cluster = consensus.cluster();
+ h.replica = consensus.replica();
+ h.nonce = nonce;
+ h.namespace = consensus.namespace();
+ h.size = size_of::<RequestStateTransferHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
+ /// `available = 0` (the requester falls back to journal repair or
+ /// retries elsewhere); an offer ships its encoded state manifest as the
+ /// frame body.
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_state_transfer_target(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ offer: Option<&metadata::StateTransferOffer>,
+ ) where
+ B: MessageBus,
+ {
+ let manifest = offer.map(|offer| consensus::encode_state_manifest(&offer.artifacts));
+ let total_size =
+ size_of::<StateTransferTargetHeader>() + manifest.as_ref().map_or(0, Vec::len);
+ let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
+ if let Some(manifest) = &manifest {
+ msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ }
+ let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
+ h.command = Command2::StateTransferTarget;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.size = total_size as u32;
+ if let Some(offer) = offer {
+ h.available = 1;
+ h.commit_op = offer.commit_op;
+ }
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_request_state_chunk(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ artifact: u32,
+ offset: u64,
+ len: u32,
+ ) where
+ B: MessageBus,
+ {
+ let msg = Message::<RequestStateChunkHeader>::new(size_of::<RequestStateChunkHeader>())
+ .transmute_header(|_, h: &mut RequestStateChunkHeader| {
+ h.command = Command2::RequestStateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.artifact = artifact;
+ h.offset = offset;
+ h.len = len;
+ h.size = size_of::<RequestStateChunkHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Serve one `RequestStateTransfer`: build a fresh offer (or refuse),
+ /// cache it for the chunk pulls, and answer with the descriptor.
+ #[allow(clippy::future_not_send)]
+ async fn on_request_state_transfer(&self, msg: &Message<RequestStateTransferHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let offer = planes.0.state_transfer_offer();
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+ if let Some(offer) = offer {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ artifacts = offer.artifacts.len(),
+ total_len = offer.artifacts.iter().map(|a| a.len).sum::<u64>(),
+ "serving metadata state transfer"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ Some(&offer),
+ )
+ .await;
+ self.metadata_transfer_offers.borrow_mut().insert(
+ header.replica,
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer,
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ } else {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "cannot serve metadata state transfer (not a caught-up \
+ primary, or no snapshot persisted); requester falls back"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ }
+
+ /// Receiver side of the descriptor: accept it and start pulling chunks,
+ /// or fall back to journal repair when the peer cannot serve.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_transfer_target(&self, msg: &Message<StateTransferTargetHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ /// Alloc cap per artifact: a corrupt length field must not OOM the
+ /// shard. Far above any real metadata snapshot or client table.
+ const ARTIFACT_LEN_MAX: u64 = 1 << 30;
+
+ /// Alloc cap across the WHOLE manifest. The per-artifact cap alone does
+ /// not bound the total: `STATE_MANIFEST_ENTRIES_MAX` allows 65k entries,
+ /// and the buffers below are reserved eagerly, so per-artifact limits
+ /// would still admit a 64 TiB reservation and abort the process. The
+ /// manifest checksum only proves it arrived intact, not that the peer
+ /// computed it sanely, so bound what this side is willing to reserve.
+ const MANIFEST_TOTAL_LEN_MAX: u64 = 4 << 30;
+
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let session_matches = self
+ .metadata_transfer
+ .borrow()
+ .as_ref()
+ .is_some_and(|session| session.nonce == header.nonce);
+ if !session_matches {
+ return;
+ }
+
+ if header.available == 0 {
+ // The peer cannot serve. If we have never installed anything the
+ // local recovery stands; run the deferred commit walk and let
+ // journal repair cover the gap (a peer that never checkpointed
+ // retains its full WAL, so repair CAN cover it).
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ "state transfer unavailable; falling back to journal repair"
+ );
+ *self.metadata_transfer.borrow_mut() = None;
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, header.replica)
+ .await;
+ return;
+ }
+
+ // The manifest rides the body; a well-formed available=1 descriptor
+ // always carries one (an empty manifest still encodes its envelope).
+ let manifest = match consensus::decode_state_manifest(
+ &msg.as_slice()[size_of::<StateTransferTargetHeader>()..header.size as usize],
+ ) {
+ Ok(manifest) => manifest,
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ %error,
+ "state transfer descriptor manifest undecodable; ignoring"
+ );
+ return;
+ }
+ };
+ if let Some(oversized) = manifest.iter().find(|entry| entry.len > ARTIFACT_LEN_MAX) {
+ tracing::error!(
+ shard = self.id,
+ kind = oversized.kind,
+ len = oversized.len,
+ "state transfer descriptor exceeds artifact cap; ignoring"
+ );
+ return;
+ }
+ let declared_total = manifest
+ .iter()
+ .fold(0u64, |total, entry| total.saturating_add(entry.len));
+ if declared_total > MANIFEST_TOTAL_LEN_MAX {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ declared_total,
+ "state transfer descriptor exceeds the total manifest cap; ignoring"
+ );
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.target_accepted {
+ // Duplicate descriptor (stall retry crossed the original).
+ return;
+ }
+ session.target_accepted = true;
+ session.commit_op = header.commit_op;
+ // Under ARTIFACT_LEN_MAX (checked above), so the casts hold.
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ session.artifacts = manifest
+ .iter()
+ .map(|&entry| ArtifactProgress {
+ entry,
+ buf: Vec::with_capacity(entry.len as usize),
+ })
+ .collect();
+ }
+ session.idle_ticks = 0;
+ }
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Fetching);
+ }
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ total_len = manifest.iter().map(|entry| entry.len).sum::<u64>(),
+ commit_op = header.commit_op,
+ "state transfer target accepted; fetching"
+ );
+ self.on_transfer_progress().await;
+ }
+
+ /// Ask for the next missing chunk of the in-flight transfer (artifacts
+ /// pulled in manifest order). No-op when nothing is missing or no
+ /// manifest is accepted yet; also the stall-retry re-request.
+ #[allow(clippy::future_not_send)]
+ async fn request_pending_state_chunk(&self)
+ where
+ B: MessageBus,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let request = {
+ let session = self.metadata_transfer.borrow();
+ session.as_ref().and_then(|session| {
+ if !session.target_accepted {
+ return None;
+ }
+ let (index, artifact) = session
+ .artifacts
+ .iter()
+ .enumerate()
+ .find(|(_, artifact)| !artifact.complete())?;
+ let offset = artifact.buf.len() as u64;
+ let remaining = artifact.entry.len - offset;
+ #[allow(clippy::cast_possible_truncation)]
+ let len = remaining.min(u64::from(STATE_CHUNK_LEN)) as u32;
+ #[allow(clippy::cast_possible_truncation)]
+ Some((session.nonce, session.peer, index as u32, offset, len))
+ })
+ };
+ if let Some((nonce, peer, artifact, offset, len)) = request {
+ self.send_request_state_chunk(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ consensus.namespace(),
+ artifact,
+ offset,
+ len,
+ )
+ .await;
+ }
+ }
+
+ /// Serve one chunk out of the cached offer. An unknown nonce (offer
+ /// evicted, e.g. the serving process restarted) answers with an
+ /// `available = 0` descriptor so the requester restarts its session.
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn on_request_state_chunk(&self, msg: &Message<RequestStateChunkHeader>)
+ where
+ B: MessageBus,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+
+ // Frame built inside the borrow; every send runs after it drops (a
+ // RefCell borrow must not cross an await on the shard).
+ // Out-of-bounds requests are dropped silently inside the block.
+ let reply = {
+ let mut offers = self.metadata_transfer_offers.borrow_mut();
+ let served = offers
+ .get_mut(&header.replica)
+ .filter(|served| served.nonce == header.nonce);
+ served.map_or(Some(ChunkReply::UnknownOffer), |served| {
+ // Serving a chunk is the only liveness signal the offer gets;
+ // the expiry sweep drops it once these stop arriving.
+ served.idle_ticks = 0;
+ // Manifest-index addressing: an index past the offer is a
+ // requester bug (or a stale frame) and is dropped below.
+ let last_artifact = served.offer.payloads.len().saturating_sub(1);
+ let artifact_bytes = served.offer.payloads.get(header.artifact as usize)?;
+ let start = header.offset as usize;
+ let end = start.saturating_add(header.len as usize);
+ // Tail of the final artifact: the receiver now holds everything
+ // the manifest promised, so the offer only has to outlive a
+ // possible re-request of this very chunk.
+ if header.artifact as usize == last_artifact && end >= artifact_bytes.len() {
+ served.fully_served = true;
+ }
+ artifact_bytes
+ .get(start..end.min(artifact_bytes.len()))
+ .map(|payload| {
+ let total_size = size_of::<StateChunkHeader>() + payload.len();
+ let mut chunk = Message::<StateChunkHeader>::new(total_size);
+ chunk.as_mut_slice()[size_of::<StateChunkHeader>()..]
+ .copy_from_slice(payload);
+ ChunkReply::Chunk(chunk.transmute_header(|_, h: &mut StateChunkHeader| {
+ h.command = Command2::StateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = header.nonce;
+ h.namespace = header.namespace;
+ h.artifact = header.artifact;
+ h.offset = header.offset;
+ h.size = total_size as u32;
+ }))
+ })
+ })
+ };
+ match reply {
+ Some(ChunkReply::Chunk(chunk)) => {
+ let _ = self
+ .bus
+ .send_to_replica(header.replica, chunk.into_generic().into_frozen())
+ .await;
+ }
+ Some(ChunkReply::UnknownOffer) => {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "state chunk request for an unknown offer; telling requester to restart"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ None => {
+ tracing::warn!(
+ shard = self.id,
+ requester = header.replica,
+ artifact = header.artifact,
+ offset = header.offset,
+ "state chunk request out of artifact bounds; ignoring"
+ );
+ }
+ }
+ }
+
+ /// Receive one chunk; on the last one, verify + install + hand the tail
+ /// to journal repair.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_chunk(&self, msg: &Message<StateChunkHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.nonce != header.nonce || !session.target_accepted {
+ return;
+ }
+ let Some(artifact) = session.artifacts.get_mut(header.artifact as usize) else {
+ return;
+ };
+ let payload = &msg.as_slice()[size_of::<StateChunkHeader>()..header.size as usize];
+ // Chunks are pulled sequentially with one in flight; anything
+ // else is a duplicate or reorder and is dropped (the stall retry
+ // re-requests from the current frontier).
+ if header.offset != artifact.buf.len() as u64 {
+ return;
+ }
+ if artifact.buf.len() as u64 + payload.len() as u64 > artifact.entry.len {
+ tracing::warn!(
+ shard = self.id,
+ artifact = header.artifact,
+ "state chunk overruns the declared artifact length; dropping frame"
+ );
+ return;
+ }
+ artifact.buf.extend_from_slice(payload);
+ session.idle_ticks = 0;
+ }
+ self.on_transfer_progress().await;
+ }
+
+ /// Drive the in-flight transfer forward: request the next missing chunk,
+ /// or - once every artifact is complete - verify, decode, and install.
+ /// Shared by descriptor acceptance and chunk arrival, so a manifest whose
+ /// artifacts are already complete (all empty) installs without waiting
+ /// for a chunk that will never come.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_transfer_progress(&self)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let complete = {
+ let session = self.metadata_transfer.borrow();
+ match session.as_ref() {
+ Some(session) if session.target_accepted => {
+ session.artifacts.iter().all(ArtifactProgress::complete)
+ }
+ _ => return,
+ }
+ };
+ if !complete {
+ self.request_pending_state_chunk().await;
+ return;
+ }
+
+ // All bytes in: verify, decode, install.
+ let session = self
+ .metadata_transfer
+ .borrow_mut()
+ .take()
+ .expect("session checked above");
+ let peer = session.peer;
+ let commit_op = session.commit_op;
+ let attempts = session.attempts;
+
+ // Per-artifact integrity, then pick the pieces this plane installs.
+ // Unknown kinds are refused rather than skipped: an artifact the
+ // serving peer thought worth shipping but this receiver cannot
+ // install would otherwise be silently dropped.
+ let mut snapshot: Option<Vec<u8>> = None;
+ let mut table: Option<(Vec<u8>, u64)> = None;
+ let mut damaged = false;
+ for (index, artifact) in session.artifacts.into_iter().enumerate() {
+ let actual = consensus::state_artifact_checksum(&artifact.buf);
+ if actual != artifact.entry.checksum {
+ tracing::error!(
+ shard = self.id,
+ artifact = index,
+ kind = artifact.entry.kind,
+ "state transfer artifact checksum mismatch"
+ );
+ damaged = true;
+ break;
+ }
+ match artifact.entry.kind {
+ consensus::artifact_kind::METADATA_SNAPSHOT => snapshot = Some(artifact.buf),
+ consensus::artifact_kind::CLIENT_TABLE => {
+ table = Some((artifact.buf, artifact.entry.frontier));
+ }
+ kind => {
+ tracing::error!(
+ shard = self.id,
+ kind,
+ "state transfer manifest carries a kind this plane cannot install"
+ );
+ damaged = true;
+ break;
+ }
+ }
+ }
+
+ let decoded = if damaged {
+ None
+ } else if let (Some(snapshot), Some((table_bytes, table_frontier))) = (snapshot, table) {
+ match consensus::ClientTable::decode(&table_bytes, self.clients_table_max.get()) {
+ Ok(table) => Some((snapshot, table, table_frontier)),
+ Err(error) => {
+ tracing::error!(shard = self.id, %error, "transferred client table undecodable");
+ None
+ }
+ }
+ } else {
+ tracing::error!(
+ shard = self.id,
+ "state transfer manifest is missing the snapshot or client table artifact"
+ );
+ None
+ };
+
+ let Some((snapshot, table, table_frontier)) = decoded else {
+ // Damage is usually transit corruption, which a re-fetch fixes. But
+ // it can also be permanent -- a peer whose artifacts this build
+ // cannot decode, or an unknown artifact kind -- and that re-offers
+ // identically every round. Carry the attempt count across the
+ // restart and share the stall budget, so a re-fetch that keeps
+ // failing gives up instead of pulling the whole snapshot forever.
+ // Distinct from the stall path: frames ARE flowing here, so
+ // `idle_ticks` never accumulates and that sweep can never fire.
+ let attempts = attempts + 1;
+ if attempts > STATE_TRANSFER_MAX_STALL_RETRIES {
+ tracing::warn!(
+ shard = self.id,
+ peer,
+ attempts,
+ "state transfer artifacts kept failing to decode; abandoning \
+ and falling back to journal repair"
+ );
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, peer).await;
+ return;
+ }
+ // Restart the session from scratch against the same peer (fresh
+ // nonce; the peer re-offers).
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::Fetching {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::AwaitingTarget);
+ }
+ let nonce = iggy_common::random_id::get_uuid();
+ *self.metadata_transfer.borrow_mut() = Some(MetadataTransferSession {
+ nonce,
+ peer,
+ commit_op: 0,
+ artifacts: Vec::new(),
+ target_accepted: false,
+ attempts,
+ idle_ticks: 0,
+ });
+ self.send_request_state_transfer(consensus, peer, nonce)
+ .await;
+ return;
+ };
+
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Installing);
Review Comment:
this `set_state_transfer_stage(Installing)` is the only unguarded stage set in the shard, and `Idle -> Installing` is an illegal transition, so it's an `assert!` panic that takes down shard 0. reachable: the probe-exhausted fallback clears the stage to `Idle` (legal from `Fetching`, so silent) but lives in the consensus crate and can't touch `metadata_transfer`, so the session survives with its nonce intact; none of the receive handlers (`on_state_transfer_target`, `on_state_chunk`, `request_pending_state_chunk`) checks the stage, so the pull completes into this assert. `on_state_transfer_target` also sets `target_accepted = true` unconditionally while moving the stage only from `AwaitingTarget` - same missing invariant from the other side. fix needs both halves: bail out of `on_transfer_progress` when the stage isn't `Fetching`, and drop the session wherever the stage gets cleared.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -1308,6 +1366,248 @@ where
Error = iggy_common::IggyError,
>,
{
+ /// Build a state-transfer offer for a restarted peer, or `None` when
+ /// this replica cannot serve one right now: not a caught-up primary
+ /// (table reads would not be authoritative), or no snapshot has ever
+ /// been persisted (the WAL then still holds the full history and the
+ /// requester's journal repair covers the whole gap).
+ ///
+ /// The snapshot is served as the on-disk bytes verbatim -- possibly
+ /// stale, which costs nothing: the receiver journal-repairs
+ /// `(snapshot_seq, commit_max]` afterwards through the existing repair
+ /// machinery. The table is encoded live at this instant; both frontier
+ /// stamps read `commit_min` inside one synchronous region, so they are
+ /// mutually consistent.
+ #[must_use]
+ pub fn state_transfer_offer(&self) -> Option<StateTransferOffer> {
+ let consensus = self.consensus.as_ref()?;
+ if !is_caught_up_primary(consensus) {
+ return None;
+ }
+ let coordinator = self.coordinator.as_ref()?;
+ let snapshot = std::fs::read(coordinator.snapshot_path()).ok()?;
Review Comment:
three problems in this function, one rewrite fixes all of them.
the raw `fs::read` ships the file including its integrity trailer, and neither end ever runs `split_trailer`: install decodes and re-persists the bytes raw, so `write_durably` appends a second trailer (+20 B per transfer generation, tolerated only because rmp-serde ignores trailing bytes - any future strict decode bricks every transferred node at boot). worse, corruption-at-rest that survives the msgpack decode (a flipped bit inside a u64 or a string body) gets re-sealed on the receiver with a fresh valid trailer and a matching pairing: a fault the source node would refuse boot over becomes undetectable on the second node.
both `.ok()?` here also swallow the error, and the shard-side refusal log then says "not a caught-up primary, or no snapshot persisted" - actively wrong when the real cause is an unreadable or corrupt snapshot.bin.
and the whole thing (read + full msgpack decode of every stream/topic/user to get one u64 + full table encode) runs synchronously on shard 0's pump, rebuilt on every `RequestStateTransfer` including stall retries.
suggested shape: `fs::read` with the error logged, `split_trailer` as verification, keep the decode for `sequence_number`, ship the stripped payload (the installed file is then byte-shape-identical to a local checkpoint), and serve repeat requests from the cached offer instead of rebuilding. deriving the seq from `last_checkpoint().0` instead does not work: `write_durably` renames before the parent-dir fsync, so a DirSync failure leaves the new file live with the cell stale, and the offer would under-advertise the frontier.
##########
core/shard/src/lib.rs:
##########
@@ -743,6 +745,96 @@ struct MetadataRepairSession {
idle_ticks: u32,
}
+/// Chunk size for state-transfer artifact pulls. Lockstep (one in flight),
+/// so the bounded per-peer bus queue can never drop a burst tail; the bus
+/// message cap is far above this.
+const STATE_CHUNK_LEN: u32 = 256 * 1024;
+
+/// Stall rounds a receiver spends on ONE peer before abandoning the transfer
+/// and falling back to journal repair. The retry has no peer re-selection, so
+/// this is what keeps a peer that died mid-transfer from wedging the rejoining
+/// node; repair then re-picks a target and re-arms a transfer if the gap is
+/// still below the new peer's retained floor.
+const STATE_TRANSFER_MAX_STALL_RETRIES: u32 = 5;
+
+/// Serving-side offer lifetime, as a multiple of the repair-retry interval. An
+/// offer resets its counter on every chunk it serves, so this only expires one
+/// that stopped being pulled -- a receiver that finished installing (the
+/// protocol has no completion frame) or gave up.
+const STATE_TRANSFER_OFFER_EXPIRY_MULTIPLE: u32 = 10;
+
+/// One artifact of an accepted transfer target: its manifest entry plus the
+/// bytes received so far (chunks are sequential, so `buf.len()` doubles as
+/// the next request offset).
+#[derive(Debug)]
+struct ArtifactProgress {
+ entry: consensus::StateArtifact,
+ buf: Vec<u8>,
+}
+
+impl ArtifactProgress {
+ const fn complete(&self) -> bool {
+ self.buf.len() as u64 == self.entry.len
+ }
+}
+
+/// One in-flight metadata state transfer (shard 0 only): a cluster-restart
+/// rejoin replacing its snapshot-shaped state (metadata snapshot + client
+/// table) from the live primary before tail repair.
+#[derive(Debug)]
+struct MetadataTransferSession {
+ nonce: u128,
+ /// Serving primary; also the stall re-request target.
+ peer: u8,
+ /// Serving peer's applied frontier from the accepted descriptor.
+ commit_op: u64,
+ /// Empty until the `StateTransferTarget` manifest is accepted, then one
+ /// entry per offered artifact, pulled in manifest order.
+ artifacts: Vec<ArtifactProgress>,
+ /// Whether a descriptor has been accepted (an accepted EMPTY manifest is
+ /// distinguishable from "still waiting").
+ target_accepted: bool,
+ /// Stall rounds this session has burned. Bounded by
+ /// [`STATE_TRANSFER_MAX_STALL_RETRIES`]: the retry always targets the same
+ /// `peer`, so a peer that dies mid-transfer would otherwise wedge the
+ /// rejoining node forever.
+ attempts: u32,
+ /// Ticks with no frame progress; at the configured repair-retry
+ /// threshold the missing piece is re-requested.
+ idle_ticks: u32,
+}
+
+/// A cached serving-side state-transfer offer (shard 0 of the serving
+/// primary). Keyed by requester replica id so a rebooted requester's fresh
+/// nonce replaces the stale offer; chunks must all come from ONE offer or
+/// the artifact checksums cannot hold.
+struct ServedStateTransfer {
Review Comment:
each cached offer owns a private copy of the snapshot plus the encoded table, keyed per requester - n-1 simultaneous rejoiners pin n-1 multi-MB copies on shard 0 until the sweep. building the offer once per snapshot generation and sharing `Rc<Vec<u8>>` payloads (the shard is single-threaded) makes the marginal requester free; `Rc` alone doesn't help while every request re-builds.
##########
core/shard/src/lib.rs:
##########
@@ -2749,6 +2964,790 @@ where
.await;
}
+ /// Start metadata tail journal-repair from `peer` when the commit walk
+ /// gap-stopped below the known frontier. Shared by `StartView` adoption
+ /// and the post-install step of a state transfer.
+ #[allow(clippy::future_not_send)]
+ async fn maybe_request_metadata_repair<P>(&self, consensus: &VsrConsensus<B, P>, peer: u8)
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ if consensus.is_normal()
+ && !consensus.is_transferring()
+ && consensus.commit_min() < consensus.commit_max()
+ && self.metadata_repair.borrow().is_none()
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ let to_op = consensus.commit_max();
+ let from_op = consensus.commit_min() + 1;
+ *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
+ nonce,
+ to_op,
+ peer,
+ idle_ticks: 0,
+ });
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ "metadata behind the group frontier; requesting repair"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
+
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn send_request_state_transfer<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ target: u8,
+ nonce: u128,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let msg =
+ Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
+ .transmute_header(|_, h: &mut RequestStateTransferHeader| {
+ h.command = Command2::RequestStateTransfer;
+ h.cluster = consensus.cluster();
+ h.replica = consensus.replica();
+ h.nonce = nonce;
+ h.namespace = consensus.namespace();
+ h.size = size_of::<RequestStateTransferHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
+ /// `available = 0` (the requester falls back to journal repair or
+ /// retries elsewhere); an offer ships its encoded state manifest as the
+ /// frame body.
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_state_transfer_target(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ offer: Option<&metadata::StateTransferOffer>,
+ ) where
+ B: MessageBus,
+ {
+ let manifest = offer.map(|offer| consensus::encode_state_manifest(&offer.artifacts));
+ let total_size =
+ size_of::<StateTransferTargetHeader>() + manifest.as_ref().map_or(0, Vec::len);
+ let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
+ if let Some(manifest) = &manifest {
+ msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ }
+ let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
+ h.command = Command2::StateTransferTarget;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.size = total_size as u32;
+ if let Some(offer) = offer {
+ h.available = 1;
+ h.commit_op = offer.commit_op;
+ }
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_request_state_chunk(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ artifact: u32,
+ offset: u64,
+ len: u32,
+ ) where
+ B: MessageBus,
+ {
+ let msg = Message::<RequestStateChunkHeader>::new(size_of::<RequestStateChunkHeader>())
+ .transmute_header(|_, h: &mut RequestStateChunkHeader| {
+ h.command = Command2::RequestStateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.artifact = artifact;
+ h.offset = offset;
+ h.len = len;
+ h.size = size_of::<RequestStateChunkHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Serve one `RequestStateTransfer`: build a fresh offer (or refuse),
+ /// cache it for the chunk pulls, and answer with the descriptor.
+ #[allow(clippy::future_not_send)]
+ async fn on_request_state_transfer(&self, msg: &Message<RequestStateTransferHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let offer = planes.0.state_transfer_offer();
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+ if let Some(offer) = offer {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ artifacts = offer.artifacts.len(),
+ total_len = offer.artifacts.iter().map(|a| a.len).sum::<u64>(),
+ "serving metadata state transfer"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ Some(&offer),
+ )
+ .await;
+ self.metadata_transfer_offers.borrow_mut().insert(
+ header.replica,
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer,
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ } else {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "cannot serve metadata state transfer (not a caught-up \
+ primary, or no snapshot persisted); requester falls back"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ }
+
+ /// Receiver side of the descriptor: accept it and start pulling chunks,
+ /// or fall back to journal repair when the peer cannot serve.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_transfer_target(&self, msg: &Message<StateTransferTargetHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ /// Alloc cap per artifact: a corrupt length field must not OOM the
+ /// shard. Far above any real metadata snapshot or client table.
+ const ARTIFACT_LEN_MAX: u64 = 1 << 30;
+
+ /// Alloc cap across the WHOLE manifest. The per-artifact cap alone does
+ /// not bound the total: `STATE_MANIFEST_ENTRIES_MAX` allows 65k entries,
+ /// and the buffers below are reserved eagerly, so per-artifact limits
+ /// would still admit a 64 TiB reservation and abort the process. The
+ /// manifest checksum only proves it arrived intact, not that the peer
+ /// computed it sanely, so bound what this side is willing to reserve.
+ const MANIFEST_TOTAL_LEN_MAX: u64 = 4 << 30;
+
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let session_matches = self
+ .metadata_transfer
+ .borrow()
+ .as_ref()
+ .is_some_and(|session| session.nonce == header.nonce);
+ if !session_matches {
+ return;
+ }
+
+ if header.available == 0 {
+ // The peer cannot serve. If we have never installed anything the
+ // local recovery stands; run the deferred commit walk and let
+ // journal repair cover the gap (a peer that never checkpointed
+ // retains its full WAL, so repair CAN cover it).
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ "state transfer unavailable; falling back to journal repair"
+ );
+ *self.metadata_transfer.borrow_mut() = None;
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, header.replica)
+ .await;
+ return;
+ }
+
+ // The manifest rides the body; a well-formed available=1 descriptor
+ // always carries one (an empty manifest still encodes its envelope).
+ let manifest = match consensus::decode_state_manifest(
+ &msg.as_slice()[size_of::<StateTransferTargetHeader>()..header.size as usize],
+ ) {
+ Ok(manifest) => manifest,
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ %error,
+ "state transfer descriptor manifest undecodable; ignoring"
+ );
+ return;
+ }
+ };
+ if let Some(oversized) = manifest.iter().find(|entry| entry.len > ARTIFACT_LEN_MAX) {
+ tracing::error!(
+ shard = self.id,
+ kind = oversized.kind,
+ len = oversized.len,
+ "state transfer descriptor exceeds artifact cap; ignoring"
+ );
+ return;
+ }
+ let declared_total = manifest
+ .iter()
+ .fold(0u64, |total, entry| total.saturating_add(entry.len));
+ if declared_total > MANIFEST_TOTAL_LEN_MAX {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ declared_total,
+ "state transfer descriptor exceeds the total manifest cap; ignoring"
+ );
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.target_accepted {
+ // Duplicate descriptor (stall retry crossed the original).
+ return;
+ }
+ session.target_accepted = true;
+ session.commit_op = header.commit_op;
+ // Under ARTIFACT_LEN_MAX (checked above), so the casts hold.
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ session.artifacts = manifest
+ .iter()
+ .map(|&entry| ArtifactProgress {
+ entry,
+ buf: Vec::with_capacity(entry.len as usize),
+ })
+ .collect();
+ }
+ session.idle_ticks = 0;
+ }
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Fetching);
+ }
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ total_len = manifest.iter().map(|entry| entry.len).sum::<u64>(),
+ commit_op = header.commit_op,
+ "state transfer target accepted; fetching"
+ );
+ self.on_transfer_progress().await;
+ }
+
+ /// Ask for the next missing chunk of the in-flight transfer (artifacts
+ /// pulled in manifest order). No-op when nothing is missing or no
+ /// manifest is accepted yet; also the stall-retry re-request.
+ #[allow(clippy::future_not_send)]
+ async fn request_pending_state_chunk(&self)
+ where
+ B: MessageBus,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let request = {
+ let session = self.metadata_transfer.borrow();
+ session.as_ref().and_then(|session| {
+ if !session.target_accepted {
+ return None;
+ }
+ let (index, artifact) = session
+ .artifacts
+ .iter()
+ .enumerate()
+ .find(|(_, artifact)| !artifact.complete())?;
+ let offset = artifact.buf.len() as u64;
+ let remaining = artifact.entry.len - offset;
+ #[allow(clippy::cast_possible_truncation)]
+ let len = remaining.min(u64::from(STATE_CHUNK_LEN)) as u32;
+ #[allow(clippy::cast_possible_truncation)]
+ Some((session.nonce, session.peer, index as u32, offset, len))
+ })
+ };
+ if let Some((nonce, peer, artifact, offset, len)) = request {
+ self.send_request_state_chunk(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ consensus.namespace(),
+ artifact,
+ offset,
+ len,
+ )
+ .await;
+ }
+ }
+
+ /// Serve one chunk out of the cached offer. An unknown nonce (offer
+ /// evicted, e.g. the serving process restarted) answers with an
+ /// `available = 0` descriptor so the requester restarts its session.
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn on_request_state_chunk(&self, msg: &Message<RequestStateChunkHeader>)
+ where
+ B: MessageBus,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+
+ // Frame built inside the borrow; every send runs after it drops (a
+ // RefCell borrow must not cross an await on the shard).
+ // Out-of-bounds requests are dropped silently inside the block.
+ let reply = {
+ let mut offers = self.metadata_transfer_offers.borrow_mut();
+ let served = offers
+ .get_mut(&header.replica)
+ .filter(|served| served.nonce == header.nonce);
+ served.map_or(Some(ChunkReply::UnknownOffer), |served| {
+ // Serving a chunk is the only liveness signal the offer gets;
+ // the expiry sweep drops it once these stop arriving.
+ served.idle_ticks = 0;
+ // Manifest-index addressing: an index past the offer is a
+ // requester bug (or a stale frame) and is dropped below.
+ let last_artifact = served.offer.payloads.len().saturating_sub(1);
+ let artifact_bytes = served.offer.payloads.get(header.artifact as usize)?;
+ let start = header.offset as usize;
+ let end = start.saturating_add(header.len as usize);
+ // Tail of the final artifact: the receiver now holds everything
+ // the manifest promised, so the offer only has to outlive a
+ // possible re-request of this very chunk.
+ if header.artifact as usize == last_artifact && end >= artifact_bytes.len() {
+ served.fully_served = true;
+ }
+ artifact_bytes
+ .get(start..end.min(artifact_bytes.len()))
+ .map(|payload| {
+ let total_size = size_of::<StateChunkHeader>() + payload.len();
+ let mut chunk = Message::<StateChunkHeader>::new(total_size);
+ chunk.as_mut_slice()[size_of::<StateChunkHeader>()..]
+ .copy_from_slice(payload);
+ ChunkReply::Chunk(chunk.transmute_header(|_, h: &mut StateChunkHeader| {
+ h.command = Command2::StateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = header.nonce;
+ h.namespace = header.namespace;
+ h.artifact = header.artifact;
+ h.offset = header.offset;
+ h.size = total_size as u32;
+ }))
+ })
+ })
+ };
+ match reply {
+ Some(ChunkReply::Chunk(chunk)) => {
+ let _ = self
+ .bus
+ .send_to_replica(header.replica, chunk.into_generic().into_frozen())
+ .await;
+ }
+ Some(ChunkReply::UnknownOffer) => {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "state chunk request for an unknown offer; telling requester to restart"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ None => {
+ tracing::warn!(
+ shard = self.id,
+ requester = header.replica,
+ artifact = header.artifact,
+ offset = header.offset,
+ "state chunk request out of artifact bounds; ignoring"
+ );
+ }
+ }
+ }
+
+ /// Receive one chunk; on the last one, verify + install + hand the tail
+ /// to journal repair.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_chunk(&self, msg: &Message<StateChunkHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.nonce != header.nonce || !session.target_accepted {
+ return;
+ }
+ let Some(artifact) = session.artifacts.get_mut(header.artifact as usize) else {
+ return;
+ };
+ let payload = &msg.as_slice()[size_of::<StateChunkHeader>()..header.size as usize];
+ // Chunks are pulled sequentially with one in flight; anything
+ // else is a duplicate or reorder and is dropped (the stall retry
+ // re-requests from the current frontier).
+ if header.offset != artifact.buf.len() as u64 {
+ return;
+ }
+ if artifact.buf.len() as u64 + payload.len() as u64 > artifact.entry.len {
+ tracing::warn!(
+ shard = self.id,
+ artifact = header.artifact,
+ "state chunk overruns the declared artifact length; dropping frame"
+ );
+ return;
+ }
+ artifact.buf.extend_from_slice(payload);
+ session.idle_ticks = 0;
+ }
+ self.on_transfer_progress().await;
+ }
+
+ /// Drive the in-flight transfer forward: request the next missing chunk,
+ /// or - once every artifact is complete - verify, decode, and install.
+ /// Shared by descriptor acceptance and chunk arrival, so a manifest whose
+ /// artifacts are already complete (all empty) installs without waiting
+ /// for a chunk that will never come.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_transfer_progress(&self)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let complete = {
+ let session = self.metadata_transfer.borrow();
+ match session.as_ref() {
+ Some(session) if session.target_accepted => {
+ session.artifacts.iter().all(ArtifactProgress::complete)
+ }
+ _ => return,
+ }
+ };
+ if !complete {
+ self.request_pending_state_chunk().await;
+ return;
+ }
+
+ // All bytes in: verify, decode, install.
+ let session = self
+ .metadata_transfer
+ .borrow_mut()
+ .take()
+ .expect("session checked above");
+ let peer = session.peer;
+ let commit_op = session.commit_op;
+ let attempts = session.attempts;
+
+ // Per-artifact integrity, then pick the pieces this plane installs.
+ // Unknown kinds are refused rather than skipped: an artifact the
+ // serving peer thought worth shipping but this receiver cannot
+ // install would otherwise be silently dropped.
+ let mut snapshot: Option<Vec<u8>> = None;
+ let mut table: Option<(Vec<u8>, u64)> = None;
+ let mut damaged = false;
+ for (index, artifact) in session.artifacts.into_iter().enumerate() {
+ let actual = consensus::state_artifact_checksum(&artifact.buf);
+ if actual != artifact.entry.checksum {
+ tracing::error!(
+ shard = self.id,
+ artifact = index,
+ kind = artifact.entry.kind,
+ "state transfer artifact checksum mismatch"
+ );
+ damaged = true;
+ break;
+ }
+ match artifact.entry.kind {
+ consensus::artifact_kind::METADATA_SNAPSHOT => snapshot = Some(artifact.buf),
+ consensus::artifact_kind::CLIENT_TABLE => {
+ table = Some((artifact.buf, artifact.entry.frontier));
+ }
+ kind => {
+ tracing::error!(
+ shard = self.id,
+ kind,
+ "state transfer manifest carries a kind this plane cannot install"
+ );
+ damaged = true;
+ break;
+ }
+ }
+ }
+
+ let decoded = if damaged {
+ None
+ } else if let (Some(snapshot), Some((table_bytes, table_frontier))) = (snapshot, table) {
+ match consensus::ClientTable::decode(&table_bytes, self.clients_table_max.get()) {
Review Comment:
decoding against the local `clients_table_max` livelocks on any capacity skew: the serving primary's live table can legitimately hold more entries than this node's configured cap - heterogeneous config, or homogeneous config after a cap reduction, since `from_snapshot` sizes capacity to `max(min_slots, highest_slot + 1)`. result is a deterministic `TooManyEntries` every round, feeding the unbounded re-arm loop above, log-only - a permanent silent join failure. decoding against the live table's capacity (needs a `capacity()` accessor) kills every route, and lets the duplicated `IggyShard::clients_table_max` cell + setter + bootstrap call go away - the same config value is already plumbed into `IggyMetadata::set_clients_table_max`.
##########
core/sdk/src/quic/quic_client.rs:
##########
@@ -177,6 +177,13 @@ impl BinaryTransport for QuicClient {
#[cfg(feature = "vsr")]
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
+ // The replayed login/register must mint a fresh Register: the failed
+ // attempt already consumed the one-shot register request id and may
+ // have half-bound the session.
+ *self
Review Comment:
this block can go: `self.disconnect()` a few lines up already calls `reset_vsr_session()`, and `begin_register` re-mints on exactly the two conditions this comment cites, on every login encode. tcp and ws reconnect paths don't have it and are fine. it's also the one session mutation that runs outside the connection lock - in the path where disconnect early-returns (retry-exhausted `connect` sets `Disconnected` without a reset), this can race an in-flight request's encode and turn it into a spurious `Unauthenticated`. deleting it restores "all session mutation happens under the connection lock".
##########
core/consensus/src/lib.rs:
##########
@@ -149,8 +149,13 @@ where
pub mod client_table;
pub use client_table::{
- CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableSnapshot,
- CommitReply,
+ CachedReply, ClientEntrySnapshot, ClientTable, ClientTableCodecError, ClientTableDecodeError,
Review Comment:
`ClientTableCodecError` next to the pre-existing `ClientTableDecodeError` - neither name says which format (msgpack checkpoint vs transfer wire), and "codec" oversells it since `encode` is infallible. `ClientTableWireError` or similar would disambiguate.
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -40,7 +40,10 @@ pub async fn pipeline_prepare_common<C, F>(
{
assert!(!consensus.is_follower(), "on_request: primary only");
assert!(consensus.is_normal(), "on_request: status must be normal");
- assert!(!consensus.is_syncing(), "on_request: must not be syncing");
+ assert!(
+ !consensus.is_transferring(),
+ "on_request: must not be syncing"
Review Comment:
assert text still says "must not be syncing" after the rename to `is_transferring` - same stale string at `metadata.rs:2759` and `iggy_partition.rs:1374` (that one's predicate changed one line above).
##########
core/shard/src/lib.rs:
##########
@@ -2563,17 +2740,55 @@ where
}
}
Command2::RangeEvicted => {
- // The metadata WAL retains everything above the snapshot
- // floor; an evicted range means the peer compacted past
- // this replica's gap -- bulk snapshot transfer (phase 3)
- // territory. Surface loudly; the divergence backstop
- // still guards the walk.
- tracing::warn!(
- shard = self.id,
- retained_from = header.op,
- "metadata repair range evicted on the serving peer; \
- snapshot transfer required"
- );
+ // Journal repair cannot close this gap: the serving peer
+ // compacted past it, so the ops this replica is missing no
+ // longer exist as WAL entries anywhere. This is the one
+ // authoritative "repair is impossible" signal, and it is
+ // shape-identical for every way a replica falls behind a
+ // checkpoint -- a fresh node joining an already-checkpointed
+ // cluster, a node whose partition healed after the quorum
+ // moved on, or a restart whose gap sits below the floor. All
+ // three convert here to state transfer against the peer that
+ // just announced the eviction (it has the checkpoint by
+ // definition), which replaces the snapshot-shaped state
+ // wholesale rather than replaying ops that are gone.
+ //
+ // Drop the repair session and arm the transfer only from
+ // `Idle`: a transfer already in flight owns the stage, and
+ // its own post-install tail repair can legitimately hit
+ // `RangeEvicted` again if the primary checkpointed mid
+ // transfer -- that reraises through the same path, and each
+ // round lifts the local floor, so it converges.
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::Idle {
+ *self.metadata_repair.borrow_mut() = None;
+ consensus.begin_state_transfer_await();
+ let nonce = iggy_common::random_id::get_uuid();
+ *self.metadata_transfer.borrow_mut() = Some(MetadataTransferSession {
+ nonce,
+ peer: header.replica,
+ commit_op: 0,
+ artifacts: Vec::new(),
+ target_accepted: false,
+ attempts: 0,
Review Comment:
the retry budget lives on the session, but three of the four writers create a fresh session, so no loop is actually bounded. this re-arm and the StartView arm both mint `attempts: 0`; only the decode-failure path carries it forward. a permanent failure (undecodable table, capacity mismatch) therefore cycles abandon -> repair -> `RangeEvicted` -> re-arm at zero forever, re-pulling the whole snapshot each round - the comment on the decode path claims the opposite. and in the other direction `attempts` is never reset on progress (chunk arrival only resets `idle_ticks`), so 5 scattered transient stalls abandon a transfer that was nearly done.
the three near-identical session literals (here, the StartView arm, the decode retry) are already drifting on exactly this field - one `arm_metadata_transfer(consensus, peer, attempts)` helper would remove the drift and make the budget a real bound (hoisted off the session, reset on progress).
##########
core/consensus/src/client_table.rs:
##########
@@ -944,6 +952,200 @@ impl ClientTable {
}
}
+/// Failure decoding an encoded client table (state transfer install path).
+#[derive(Debug)]
+pub enum ClientTableCodecError {
+ /// Byte stream ended mid-field.
+ Truncated,
+ /// Leading magic is not [`CLIENT_TABLE_MAGIC`].
+ BadMagic,
+ /// Trailing hash does not match the content.
+ ChecksumMismatch { expected: u64, actual: u64 },
+ /// Encoded entry count exceeds this table's capacity.
+ TooManyEntries { count: u32, max: usize },
+ /// A cached reply's bytes do not parse as a valid reply message.
+ InvalidReply,
+ /// An entry carries an empty reply ring (violates the never-empty
+ /// invariant registration establishes).
+ EmptyRing,
+}
+
+impl std::fmt::Display for ClientTableCodecError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Truncated => write!(f, "encoded client table truncated"),
+ Self::BadMagic => write!(f, "encoded client table has wrong magic"),
+ Self::ChecksumMismatch { expected, actual } => write!(
+ f,
+ "client table checksum mismatch: expected {expected:#018x}, actual {actual:#018x}"
+ ),
+ Self::TooManyEntries { count, max } => {
+ write!(f, "encoded client table holds {count} entries, max {max}")
+ }
+ Self::InvalidReply => write!(f, "encoded client table holds an invalid cached reply"),
+ Self::EmptyRing => write!(f, "encoded client table entry has an empty reply ring"),
+ }
+ }
+}
+
+impl std::error::Error for ClientTableCodecError {}
+
+/// Format tag for [`ClientTable::encode`]; bump on layout change.
+pub const CLIENT_TABLE_MAGIC: [u8; 4] = *b"ICT1";
+
+impl ClientTable {
+ /// Encode the table for state transfer.
+ ///
+ /// Layout (all little-endian): `magic(4) count(u32)` then per entry in
+ /// slot order `client(u128) epoch(u64) user_id(u32) watermark(u64)
+ /// watermark_checksum(u128) ring_len(u8) [reply_len(u32) reply_bytes]*`,
+ /// terminated by an `XxHash3_64(8)` over everything before it.
+ ///
+ /// Slot order makes the bytes deterministic across caught-up replicas:
+ /// slots are apply-derived state.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn encode(&self) -> Vec<u8> {
+ let mut out = Vec::with_capacity(64 * self.index.len() + 16);
+ out.extend_from_slice(&CLIENT_TABLE_MAGIC);
+ out.extend_from_slice(&(self.index.len() as u32).to_le_bytes());
+ for (slot_idx, slot) in self.slots.iter().enumerate() {
+ let Some(entry) = slot else { continue };
+ debug_assert_eq!(self.index.get(&entry.client_id), Some(&slot_idx));
+ out.extend_from_slice(&entry.client_id.to_le_bytes());
+ out.extend_from_slice(&entry.epoch.to_le_bytes());
+ out.extend_from_slice(&entry.user_id.to_le_bytes());
+ out.extend_from_slice(&entry.watermark.to_le_bytes());
+ out.extend_from_slice(&entry.watermark_checksum.to_le_bytes());
+ out.push(entry.ring.len() as u8);
+ for reply in &entry.ring {
+ let bytes = reply.bytes.as_slice();
+ out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
+ out.extend_from_slice(bytes);
+ }
+ }
+ let mut hasher = XxHash3_64::new();
+ hasher.write(&out);
+ out.extend_from_slice(&hasher.finish().to_le_bytes());
+ out
+ }
+
+ /// Decode a table encoded by [`Self::encode`] into a fresh table of
+ /// `max_clients` capacity.
+ ///
+ /// The denormalized `latest_commit` is rebuilt from the decoded ring's
+ /// back, not trusted from the wire, so it cannot drift from the ring.
+ ///
+ /// # Errors
+ /// [`ClientTableCodecError`] on truncation, magic/checksum mismatch,
+ /// capacity overflow, or an undecodable cached reply.
+ ///
+ /// # Panics
+ /// Unreachable: slice-to-array conversions are length-checked first.
+ pub fn decode(bytes: &[u8], max_clients: usize) -> Result<Self, ClientTableCodecError> {
+ const TRAILER: usize = size_of::<u64>();
+ let content_len = bytes
+ .len()
+ .checked_sub(TRAILER)
+ .ok_or(ClientTableCodecError::Truncated)?;
+ let (content, trailer) = bytes.split_at(content_len);
+ let expected = u64::from_le_bytes(trailer.try_into().expect("trailer is 8 bytes"));
+ let mut hasher = XxHash3_64::new();
+ hasher.write(content);
+ let actual = hasher.finish();
+ if expected != actual {
+ return Err(ClientTableCodecError::ChecksumMismatch { expected, actual });
+ }
+
+ let mut reader = CodecReader { bytes: content };
+ if reader.take(CLIENT_TABLE_MAGIC.len())? != CLIENT_TABLE_MAGIC {
+ return Err(ClientTableCodecError::BadMagic);
+ }
+ let count = reader.u32()?;
+ if count as usize > max_clients {
+ return Err(ClientTableCodecError::TooManyEntries {
+ count,
+ max: max_clients,
+ });
+ }
+
+ let mut table = Self::new(max_clients);
+ for slot_idx in 0..count as usize {
+ let client_id = reader.u128()?;
+ let epoch = reader.u64()?;
+ let user_id = reader.u32()?;
+ let watermark = reader.u64()?;
+ let watermark_checksum = reader.u128()?;
+ let ring_len = reader.u8()?;
+ if ring_len == 0 {
+ return Err(ClientTableCodecError::EmptyRing);
+ }
+ let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY);
+ for _ in 0..ring_len {
+ let reply_len = reader.u32()? as usize;
+ let reply_bytes = reader.take(reply_len)?;
+ let owned = Owned::<MESSAGE_ALIGN>::copy_from_slice(reply_bytes);
+ let message = Message::<GenericHeader>::try_from(owned)
+ .map_err(|_| ClientTableCodecError::InvalidReply)?
+ .try_into_typed::<ReplyHeader>()
+ .map_err(|_| ClientTableCodecError::InvalidReply)?;
+ ring.push_back(CachedReply::from_message(message));
+ }
+ let latest_commit = ring
+ .back()
+ .expect("ring_len checked non-zero above")
+ .header()
+ .commit;
+ table.slots[slot_idx] = Some(ClientEntry {
+ epoch,
+ user_id,
+ watermark,
+ watermark_checksum,
+ ring,
+ client_id,
+ latest_commit,
+ });
+ table.index.insert(client_id, slot_idx);
Review Comment:
the wire decode skips three validations `from_snapshot` treats as mandatory, on strictly less trusted input. duplicate `client_id`: this `index.insert` silently overwrites, leaving an occupied-but-unindexed slot - with `count == max_clients` plus one duplicate, `commit_register` later sees `index.len() >= slots.len()` as false (skips eviction), `first_free_slot()` finds nothing, and the `.expect("eviction must free a slot")` panics shard 0. `ring_len`: accepted up to 255 with no `REPLY_RING_CAPACITY` bound - `push_latest` only evicts on `==`, so an over-capacity ring grows by one on every later `commit_reply` for that client, and `encode`'s `ring.len() as u8` wraps at 256, making the table permanently un-transferable. the artifact checksum only proves transit, so a corrupt or buggy peer's table reaches this codepath - exactly the threat model the sibling decoder documents (`DuplicateClientId`, `DuplicateSlot`, `SlotOutOfRange`). worth porting all three bounds.
separately: entries compact into `0..count` here, so the encode doc's "every caught-up replica encodes identical bytes" stops being true after one transfer (new registrations land in different slots). eviction victims are unaffected, so just fix the doc.
##########
core/journal/src/lib.rs:
##########
@@ -60,6 +60,17 @@ where
) -> impl Future<Output = io::Result<Vec<Self::Entry>>> {
async { Ok(Vec::new()) }
}
+
+ /// Snapshot watermark: entries at or below it are evictable. `0` for
+ /// journals without snapshot bookkeeping.
+ fn snapshot_op(&self) -> u64 {
Review Comment:
these defaults are a trap through `JournalState<J>`, which forwards only `header`/`previous_header`/`append`/`entry` - through the wrapper `set_snapshot_op` is a silent no-op and `snapshot_op()` returns 0 (the pre-existing `remaining_capacity` default already makes `should_checkpoint` permanently false there too). latent today because the metadata plane holds `PrepareJournal` directly, but the moment partition-plane transfer lands on the wrapper this breaks silently - or panics, since forwarding `set_snapshot_op` while inheriting the `snapshot_op() -> 0` default turns the install's monotonicity guard into an assert trip. making both methods required (or forwarding everything in `JournalState`) closes it in both directions.
##########
core/shard/src/lib.rs:
##########
@@ -2749,6 +2964,790 @@ where
.await;
}
+ /// Start metadata tail journal-repair from `peer` when the commit walk
+ /// gap-stopped below the known frontier. Shared by `StartView` adoption
+ /// and the post-install step of a state transfer.
+ #[allow(clippy::future_not_send)]
+ async fn maybe_request_metadata_repair<P>(&self, consensus: &VsrConsensus<B, P>, peer: u8)
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ if consensus.is_normal()
+ && !consensus.is_transferring()
+ && consensus.commit_min() < consensus.commit_max()
+ && self.metadata_repair.borrow().is_none()
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ let to_op = consensus.commit_max();
+ let from_op = consensus.commit_min() + 1;
+ *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
+ nonce,
+ to_op,
+ peer,
+ idle_ticks: 0,
+ });
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ "metadata behind the group frontier; requesting repair"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
+
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn send_request_state_transfer<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ target: u8,
+ nonce: u128,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let msg =
+ Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
+ .transmute_header(|_, h: &mut RequestStateTransferHeader| {
+ h.command = Command2::RequestStateTransfer;
+ h.cluster = consensus.cluster();
+ h.replica = consensus.replica();
+ h.nonce = nonce;
+ h.namespace = consensus.namespace();
+ h.size = size_of::<RequestStateTransferHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
+ /// `available = 0` (the requester falls back to journal repair or
+ /// retries elsewhere); an offer ships its encoded state manifest as the
+ /// frame body.
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_state_transfer_target(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ offer: Option<&metadata::StateTransferOffer>,
+ ) where
+ B: MessageBus,
+ {
+ let manifest = offer.map(|offer| consensus::encode_state_manifest(&offer.artifacts));
+ let total_size =
+ size_of::<StateTransferTargetHeader>() + manifest.as_ref().map_or(0, Vec::len);
+ let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
+ if let Some(manifest) = &manifest {
+ msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ }
+ let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
+ h.command = Command2::StateTransferTarget;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.size = total_size as u32;
+ if let Some(offer) = offer {
+ h.available = 1;
+ h.commit_op = offer.commit_op;
+ }
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_request_state_chunk(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ artifact: u32,
+ offset: u64,
+ len: u32,
+ ) where
+ B: MessageBus,
+ {
+ let msg = Message::<RequestStateChunkHeader>::new(size_of::<RequestStateChunkHeader>())
+ .transmute_header(|_, h: &mut RequestStateChunkHeader| {
+ h.command = Command2::RequestStateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.artifact = artifact;
+ h.offset = offset;
+ h.len = len;
+ h.size = size_of::<RequestStateChunkHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Serve one `RequestStateTransfer`: build a fresh offer (or refuse),
+ /// cache it for the chunk pulls, and answer with the descriptor.
+ #[allow(clippy::future_not_send)]
+ async fn on_request_state_transfer(&self, msg: &Message<RequestStateTransferHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let offer = planes.0.state_transfer_offer();
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+ if let Some(offer) = offer {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ artifacts = offer.artifacts.len(),
+ total_len = offer.artifacts.iter().map(|a| a.len).sum::<u64>(),
+ "serving metadata state transfer"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ Some(&offer),
+ )
+ .await;
+ self.metadata_transfer_offers.borrow_mut().insert(
+ header.replica,
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer,
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ } else {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "cannot serve metadata state transfer (not a caught-up \
+ primary, or no snapshot persisted); requester falls back"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ }
+
+ /// Receiver side of the descriptor: accept it and start pulling chunks,
+ /// or fall back to journal repair when the peer cannot serve.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_transfer_target(&self, msg: &Message<StateTransferTargetHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ /// Alloc cap per artifact: a corrupt length field must not OOM the
+ /// shard. Far above any real metadata snapshot or client table.
+ const ARTIFACT_LEN_MAX: u64 = 1 << 30;
+
+ /// Alloc cap across the WHOLE manifest. The per-artifact cap alone does
+ /// not bound the total: `STATE_MANIFEST_ENTRIES_MAX` allows 65k entries,
+ /// and the buffers below are reserved eagerly, so per-artifact limits
+ /// would still admit a 64 TiB reservation and abort the process. The
+ /// manifest checksum only proves it arrived intact, not that the peer
+ /// computed it sanely, so bound what this side is willing to reserve.
+ const MANIFEST_TOTAL_LEN_MAX: u64 = 4 << 30;
+
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let session_matches = self
+ .metadata_transfer
+ .borrow()
+ .as_ref()
+ .is_some_and(|session| session.nonce == header.nonce);
+ if !session_matches {
+ return;
+ }
+
+ if header.available == 0 {
+ // The peer cannot serve. If we have never installed anything the
+ // local recovery stands; run the deferred commit walk and let
+ // journal repair cover the gap (a peer that never checkpointed
+ // retains its full WAL, so repair CAN cover it).
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ "state transfer unavailable; falling back to journal repair"
+ );
+ *self.metadata_transfer.borrow_mut() = None;
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, header.replica)
+ .await;
+ return;
+ }
+
+ // The manifest rides the body; a well-formed available=1 descriptor
+ // always carries one (an empty manifest still encodes its envelope).
+ let manifest = match consensus::decode_state_manifest(
+ &msg.as_slice()[size_of::<StateTransferTargetHeader>()..header.size as usize],
+ ) {
+ Ok(manifest) => manifest,
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ %error,
+ "state transfer descriptor manifest undecodable; ignoring"
+ );
+ return;
+ }
+ };
+ if let Some(oversized) = manifest.iter().find(|entry| entry.len > ARTIFACT_LEN_MAX) {
+ tracing::error!(
+ shard = self.id,
+ kind = oversized.kind,
+ len = oversized.len,
+ "state transfer descriptor exceeds artifact cap; ignoring"
+ );
+ return;
+ }
+ let declared_total = manifest
+ .iter()
+ .fold(0u64, |total, entry| total.saturating_add(entry.len));
+ if declared_total > MANIFEST_TOTAL_LEN_MAX {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ declared_total,
+ "state transfer descriptor exceeds the total manifest cap; ignoring"
+ );
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.target_accepted {
+ // Duplicate descriptor (stall retry crossed the original).
+ return;
+ }
+ session.target_accepted = true;
+ session.commit_op = header.commit_op;
+ // Under ARTIFACT_LEN_MAX (checked above), so the casts hold.
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ session.artifacts = manifest
+ .iter()
+ .map(|&entry| ArtifactProgress {
+ entry,
+ buf: Vec::with_capacity(entry.len as usize),
+ })
+ .collect();
+ }
+ session.idle_ticks = 0;
+ }
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Fetching);
+ }
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ total_len = manifest.iter().map(|entry| entry.len).sum::<u64>(),
+ commit_op = header.commit_op,
+ "state transfer target accepted; fetching"
+ );
+ self.on_transfer_progress().await;
+ }
+
+ /// Ask for the next missing chunk of the in-flight transfer (artifacts
+ /// pulled in manifest order). No-op when nothing is missing or no
+ /// manifest is accepted yet; also the stall-retry re-request.
+ #[allow(clippy::future_not_send)]
+ async fn request_pending_state_chunk(&self)
+ where
+ B: MessageBus,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let request = {
+ let session = self.metadata_transfer.borrow();
+ session.as_ref().and_then(|session| {
+ if !session.target_accepted {
+ return None;
+ }
+ let (index, artifact) = session
+ .artifacts
+ .iter()
+ .enumerate()
+ .find(|(_, artifact)| !artifact.complete())?;
+ let offset = artifact.buf.len() as u64;
+ let remaining = artifact.entry.len - offset;
+ #[allow(clippy::cast_possible_truncation)]
+ let len = remaining.min(u64::from(STATE_CHUNK_LEN)) as u32;
+ #[allow(clippy::cast_possible_truncation)]
+ Some((session.nonce, session.peer, index as u32, offset, len))
+ })
+ };
+ if let Some((nonce, peer, artifact, offset, len)) = request {
+ self.send_request_state_chunk(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ consensus.namespace(),
+ artifact,
+ offset,
+ len,
+ )
+ .await;
+ }
+ }
+
+ /// Serve one chunk out of the cached offer. An unknown nonce (offer
+ /// evicted, e.g. the serving process restarted) answers with an
+ /// `available = 0` descriptor so the requester restarts its session.
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn on_request_state_chunk(&self, msg: &Message<RequestStateChunkHeader>)
+ where
+ B: MessageBus,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+
+ // Frame built inside the borrow; every send runs after it drops (a
+ // RefCell borrow must not cross an await on the shard).
+ // Out-of-bounds requests are dropped silently inside the block.
+ let reply = {
+ let mut offers = self.metadata_transfer_offers.borrow_mut();
+ let served = offers
+ .get_mut(&header.replica)
+ .filter(|served| served.nonce == header.nonce);
+ served.map_or(Some(ChunkReply::UnknownOffer), |served| {
+ // Serving a chunk is the only liveness signal the offer gets;
+ // the expiry sweep drops it once these stop arriving.
+ served.idle_ticks = 0;
+ // Manifest-index addressing: an index past the offer is a
+ // requester bug (or a stale frame) and is dropped below.
+ let last_artifact = served.offer.payloads.len().saturating_sub(1);
+ let artifact_bytes = served.offer.payloads.get(header.artifact as usize)?;
+ let start = header.offset as usize;
+ let end = start.saturating_add(header.len as usize);
+ // Tail of the final artifact: the receiver now holds everything
+ // the manifest promised, so the offer only has to outlive a
+ // possible re-request of this very chunk.
+ if header.artifact as usize == last_artifact && end >= artifact_bytes.len() {
+ served.fully_served = true;
+ }
+ artifact_bytes
+ .get(start..end.min(artifact_bytes.len()))
Review Comment:
`get(start..end.min(artifact_bytes.len()))` returns `Some(&[])` when `start == len`, so this answers a fully-consumed artifact with an empty chunk instead of dropping the request - and the handler already reset `served.idle_ticks` above, so the offer never expires. the receiver accepts the empty payload (offset matches, no overrun), extends nothing, resets its own `idle_ticks`, and immediately re-requests the same offset. both liveness counters pinned at zero, `attempts` never increments: the transfer has no bound at all, the two nodes ping-pong empty frames at network rate, and the rejoining replica withholds `PrepareOk` for the life of the process - a 3-node cluster is permanently down to 2 acking replicas.
reachable via the offer rebuild: a stall-retry `RequestStateTransfer` reuses the session nonce, and `on_request_state_transfer` rebuilds the offer and replaces the cache entry under that same nonce. the client table is encoded live, so if the first descriptor was delayed rather than lost, the receiver holds manifest A while chunks come from offer B. `B.len > A.len` is caught by the artifact checksum; `B.len < A.len` (a client logged out between the two builds) lands the cursor exactly on `B.len` and enters this loop.
fix wants all three: refuse chunk requests with `start >= len`, don't reset either idle counter on a zero-byte payload, and make the cached offer first-wins per `(requester, nonce)` so a rebuild can't replace a manifest the receiver already accepted.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -1308,6 +1366,248 @@ where
Error = iggy_common::IggyError,
>,
{
+ /// Build a state-transfer offer for a restarted peer, or `None` when
+ /// this replica cannot serve one right now: not a caught-up primary
+ /// (table reads would not be authoritative), or no snapshot has ever
+ /// been persisted (the WAL then still holds the full history and the
+ /// requester's journal repair covers the whole gap).
+ ///
+ /// The snapshot is served as the on-disk bytes verbatim -- possibly
+ /// stale, which costs nothing: the receiver journal-repairs
+ /// `(snapshot_seq, commit_max]` afterwards through the existing repair
+ /// machinery. The table is encoded live at this instant; both frontier
+ /// stamps read `commit_min` inside one synchronous region, so they are
+ /// mutually consistent.
+ #[must_use]
+ pub fn state_transfer_offer(&self) -> Option<StateTransferOffer> {
+ let consensus = self.consensus.as_ref()?;
+ if !is_caught_up_primary(consensus) {
+ return None;
+ }
+ let coordinator = self.coordinator.as_ref()?;
+ let snapshot = std::fs::read(coordinator.snapshot_path()).ok()?;
+ let snapshot_seq = IggySnapshot::decode(&snapshot).ok()?.sequence_number();
+ let commit_op = consensus.commit_min();
+ let table = self.client_table.borrow().encode();
+ Some(StateTransferOffer {
+ commit_op,
+ artifacts: vec![
+ consensus::StateArtifact::for_bytes(
+ consensus::artifact_kind::METADATA_SNAPSHOT,
+ snapshot_seq,
+ &snapshot,
+ ),
+ consensus::StateArtifact::for_bytes(
+ consensus::artifact_kind::CLIENT_TABLE,
+ commit_op,
+ &table,
+ ),
+ ],
+ payloads: vec![snapshot, table],
+ })
+ }
+
+ /// Install a fetched state transfer: persist + restore the snapshot,
+ /// replace the client table, and jump the commit state to the snapshot
+ /// floor so the tail repair takes over from there.
+ ///
+ /// Ordering: persist FIRST (a crash mid-install must reboot from the
+ /// transferred state, not the pre-transfer one), then the in-place STM
+ /// restore (readers observe it on their next read), then the table +
+ /// frontier, then journal/commit bookkeeping.
+ ///
+ /// The partition plane is deliberately untouched: partitions load
+ /// whatever their disks hold at boot and repair through their own
+ /// consensus groups. Topology changes the snapshot carries below the
+ /// receiver's old frontier (topics created/deleted while it was down)
+ /// fire no commit notifier -- convergence rests on the partition
+ /// reconciler's periodic full diff against the committed STM, which
+ /// reads the restored state on its next tick.
+ ///
+ /// Returns the installed snapshot sequence (the receiver's new applied
+ /// frontier).
+ ///
+ /// # Errors
+ /// [`SnapshotError`] when the snapshot bytes do not decode, the persist
+ /// fails, or the in-place restore is rejected.
+ ///
+ /// # Panics
+ /// If called on a shard without consensus (state transfer is a shard-0
+ /// concern).
+ #[allow(clippy::future_not_send)]
+ pub async fn install_state_transfer(
+ &self,
+ snapshot_bytes: &[u8],
+ client_table: ClientTable,
+ table_frontier: u64,
+ commit_op: u64,
+ ) -> Result<u64, SnapshotError>
+ where
+ M: RestoreSnapshotInPlace<MetadataSnapshot>,
+ {
+ let consensus = self
+ .consensus
+ .as_ref()
+ .expect("install_state_transfer: consensus only exists on shard 0");
+
+ let snapshot = IggySnapshot::decode(snapshot_bytes)?;
+ let snapshot_seq = snapshot.sequence_number();
+
+ // Manifest coherence. `commit_op` and `table_frontier` arrive from the
+ // serving peer and are applied to THIS replica's frontiers, so a
+ // malformed descriptor would move them somewhere the artifacts do not
+ // justify. A peer cannot have committed less than its own snapshot
+ // contains, nor have encoded a table below its commit point: both are
+ // built from one caught-up-primary read in `state_transfer_offer`.
+ // Refuse rather than install, which drops the caller back to journal
+ // repair with the local state untouched.
+ if commit_op < snapshot_seq || table_frontier > commit_op {
+ tracing::error!(
+ snapshot_seq,
+ commit_op,
+ table_frontier,
+ "incoherent state transfer manifest; refusing to install"
+ );
+ return Err(SnapshotError::IncoherentManifest {
+ snapshot_seq,
+ commit_op,
+ table_frontier,
+ });
+ }
+
+ // Checkpoints are node-local, so a healthy serving primary can offer
+ // a snapshot BEHIND this replica's own applied frontier (each node
+ // snapshots at its own watermark; a backup checkpoints an op or two
+ // below the primary it later replaces). Restoring such a snapshot
+ // would rewind the STM below `commit_min` with no way back: the
+ // commit walk never revisits ops it already counted as applied, so
+ // the rewound-over effects would be lost until the next transfer.
+ // Keep the local STM (it is a superset) and let tail repair cover
+ // `(commit_min, commit_op]`. The client table still installs below:
+ // it comes from the serving primary's LIVE state at `table_frontier
+ // == commit_op`, which is never behind this replica.
+ let local_applied = consensus.commit_min();
+ let snapshot_ahead = snapshot_seq > local_applied;
+
+ // Hold the superblock gate across the ENTIRE install, not just its
+ // superblock write. A checkpoint does the same pair of steps -- rewrite
+ // `snapshot.bin`, record `(checkpoint_op, checksum)` -- so interleaving
+ // the two could leave the file from one and the durable pairing from the
+ // other, which is precisely the torn pairing the superblock exists to
+ // detect. Serializing against `checkpoint_if_needed` makes that
+ // impossible by construction rather than by argument (a transferring
+ // replica withholds acks, so it should not be committing, but "should"
+ // is not an invariant this path can rely on).
+ //
+ // Deadlock-free: nothing between here and the write awaits, and
+ // `write_superblock` takes no lock of its own -- the checkpoint path
+ // calls it under this same gate.
+ let _superblock_gate = if snapshot_ahead && self.superblock.is_some() {
+ Some(self.superblock_lock.acquire().await)
+ } else {
+ None
+ };
+
+ if snapshot_ahead {
+ if let Some(coordinator) = &self.coordinator {
+ // The transferred snapshot REPLACES the one the superblock's
+ // `(checkpoint_op, checksum)` pairing describes, so the pairing has
+ // to move with it. Left stale it does not refuse boot -- the
+ // `checkpoint_op < snapshot_op` arm of `verify_checkpoint_pairing`
+ // reads it as a lagging local checkpoint and accepts -- which is
+ // worse than a refusal: the recorded checksum belongs to a snapshot
+ // that no longer exists, so a torn or corrupt transferred snapshot
+ // stops being detectable until some later local checkpoint happens
+ // to rewrite the pairing.
+ //
+ // Write the received bytes verbatim rather than re-encoding the
+ // decoded snapshot: the checksum below is taken over exactly the
+ // bytes that reach the file, so the pairing provably describes it.
+ debug_assert!(
+ snapshot_seq >= coordinator.last_checkpoint().0,
+ "a transferred snapshot must not land below the recorded \
+ checkpoint op ({} < {}); recovery would refuse boot with \
+ CheckpointAheadOfSnapshot",
+ snapshot_seq,
+ coordinator.last_checkpoint().0
+ );
+ let checksum = checkpoint_checksum(snapshot_bytes);
+ IggySnapshot::write_durably(&coordinator.snapshot_path(), snapshot_bytes)?;
+ coordinator.seed_last_checkpoint(snapshot_seq, checksum);
+ tracing::info!(
+ checkpoint_op = snapshot_seq,
+ "state transfer recorded its checkpoint pairing"
+ );
+ } else {
+ tracing::warn!(
+ snapshot_seq,
+ "installing state transfer without a snapshot coordinator; \
+ the transferred state will not survive a further restart"
+ );
+ }
+
+ self.mux_stm
+ .restore_snapshot_in_place(snapshot.snapshot())?;
+ } else {
+ tracing::info!(
+ snapshot_seq,
+ local_applied,
+ "transferred snapshot at or below the local applied frontier; \
+ keeping the local state machine and installing the table only"
+ );
+ }
+
+ *self.client_table.borrow_mut() = client_table;
+ self.client_table_frontier.set(table_frontier);
+
+ if snapshot_ahead {
+ // Entries at or below the installed floor are superseded by the
+ // snapshot; without this the journal's wrap-eviction assert trips
+ // on pre-transfer residents the next time slots recycle.
+ if let Some(journal) = &self.journal {
+ let handle = journal.handle();
+ if snapshot_seq > handle.snapshot_op() {
+ handle.set_snapshot_op(snapshot_seq);
+ }
+ }
+
+ // The snapshot IS ops `..=snapshot_seq` applied: jump the applied
+ // frontier (this is the op-jump the tail repair resumes from) and
+ // let the announced commit point pull the walk target forward.
+ consensus.set_commit_floor(snapshot_seq);
Review Comment:
flagging as a follow-up rather than for this PR: `set_commit_floor(snapshot_seq)` hands the commit walk a floor it never verified, and the walk matches entries by op number only - no view or hash-chain check - while `set_snapshot_op` leaves everything above the floor resident. a node with a pre-crash prepared-but-uncommitted suffix that a view change reassigned cluster-side would apply those stale bodies as committed (the table is shielded by the frontier fence, the STM is not). the `TODO(suffix-truncation)` in consensus names the same root cause and the missing durable truncate-from-op primitive; this floor jump widens exposure to it on exactly the nodes guaranteed to have a stale log.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -1308,6 +1366,248 @@ where
Error = iggy_common::IggyError,
>,
{
+ /// Build a state-transfer offer for a restarted peer, or `None` when
+ /// this replica cannot serve one right now: not a caught-up primary
+ /// (table reads would not be authoritative), or no snapshot has ever
+ /// been persisted (the WAL then still holds the full history and the
+ /// requester's journal repair covers the whole gap).
+ ///
+ /// The snapshot is served as the on-disk bytes verbatim -- possibly
+ /// stale, which costs nothing: the receiver journal-repairs
+ /// `(snapshot_seq, commit_max]` afterwards through the existing repair
+ /// machinery. The table is encoded live at this instant; both frontier
+ /// stamps read `commit_min` inside one synchronous region, so they are
+ /// mutually consistent.
+ #[must_use]
+ pub fn state_transfer_offer(&self) -> Option<StateTransferOffer> {
+ let consensus = self.consensus.as_ref()?;
+ if !is_caught_up_primary(consensus) {
+ return None;
+ }
+ let coordinator = self.coordinator.as_ref()?;
+ let snapshot = std::fs::read(coordinator.snapshot_path()).ok()?;
+ let snapshot_seq = IggySnapshot::decode(&snapshot).ok()?.sequence_number();
+ let commit_op = consensus.commit_min();
+ let table = self.client_table.borrow().encode();
+ Some(StateTransferOffer {
+ commit_op,
+ artifacts: vec![
+ consensus::StateArtifact::for_bytes(
+ consensus::artifact_kind::METADATA_SNAPSHOT,
+ snapshot_seq,
+ &snapshot,
+ ),
+ consensus::StateArtifact::for_bytes(
+ consensus::artifact_kind::CLIENT_TABLE,
+ commit_op,
+ &table,
+ ),
+ ],
+ payloads: vec![snapshot, table],
+ })
+ }
+
+ /// Install a fetched state transfer: persist + restore the snapshot,
+ /// replace the client table, and jump the commit state to the snapshot
+ /// floor so the tail repair takes over from there.
+ ///
+ /// Ordering: persist FIRST (a crash mid-install must reboot from the
+ /// transferred state, not the pre-transfer one), then the in-place STM
+ /// restore (readers observe it on their next read), then the table +
+ /// frontier, then journal/commit bookkeeping.
+ ///
+ /// The partition plane is deliberately untouched: partitions load
+ /// whatever their disks hold at boot and repair through their own
+ /// consensus groups. Topology changes the snapshot carries below the
+ /// receiver's old frontier (topics created/deleted while it was down)
+ /// fire no commit notifier -- convergence rests on the partition
+ /// reconciler's periodic full diff against the committed STM, which
+ /// reads the restored state on its next tick.
+ ///
+ /// Returns the installed snapshot sequence (the receiver's new applied
+ /// frontier).
+ ///
+ /// # Errors
+ /// [`SnapshotError`] when the snapshot bytes do not decode, the persist
+ /// fails, or the in-place restore is rejected.
+ ///
+ /// # Panics
+ /// If called on a shard without consensus (state transfer is a shard-0
+ /// concern).
+ #[allow(clippy::future_not_send)]
+ pub async fn install_state_transfer(
+ &self,
+ snapshot_bytes: &[u8],
+ client_table: ClientTable,
+ table_frontier: u64,
+ commit_op: u64,
+ ) -> Result<u64, SnapshotError>
+ where
+ M: RestoreSnapshotInPlace<MetadataSnapshot>,
+ {
+ let consensus = self
+ .consensus
+ .as_ref()
+ .expect("install_state_transfer: consensus only exists on shard 0");
+
+ let snapshot = IggySnapshot::decode(snapshot_bytes)?;
+ let snapshot_seq = snapshot.sequence_number();
+
+ // Manifest coherence. `commit_op` and `table_frontier` arrive from the
+ // serving peer and are applied to THIS replica's frontiers, so a
+ // malformed descriptor would move them somewhere the artifacts do not
+ // justify. A peer cannot have committed less than its own snapshot
+ // contains, nor have encoded a table below its commit point: both are
+ // built from one caught-up-primary read in `state_transfer_offer`.
+ // Refuse rather than install, which drops the caller back to journal
+ // repair with the local state untouched.
+ if commit_op < snapshot_seq || table_frontier > commit_op {
+ tracing::error!(
+ snapshot_seq,
+ commit_op,
+ table_frontier,
+ "incoherent state transfer manifest; refusing to install"
+ );
+ return Err(SnapshotError::IncoherentManifest {
+ snapshot_seq,
+ commit_op,
+ table_frontier,
+ });
+ }
+
+ // Checkpoints are node-local, so a healthy serving primary can offer
+ // a snapshot BEHIND this replica's own applied frontier (each node
+ // snapshots at its own watermark; a backup checkpoints an op or two
+ // below the primary it later replaces). Restoring such a snapshot
+ // would rewind the STM below `commit_min` with no way back: the
+ // commit walk never revisits ops it already counted as applied, so
+ // the rewound-over effects would be lost until the next transfer.
+ // Keep the local STM (it is a superset) and let tail repair cover
+ // `(commit_min, commit_op]`. The client table still installs below:
+ // it comes from the serving primary's LIVE state at `table_frontier
+ // == commit_op`, which is never behind this replica.
+ let local_applied = consensus.commit_min();
+ let snapshot_ahead = snapshot_seq > local_applied;
+
+ // Hold the superblock gate across the ENTIRE install, not just its
+ // superblock write. A checkpoint does the same pair of steps -- rewrite
+ // `snapshot.bin`, record `(checkpoint_op, checksum)` -- so interleaving
+ // the two could leave the file from one and the durable pairing from the
+ // other, which is precisely the torn pairing the superblock exists to
+ // detect. Serializing against `checkpoint_if_needed` makes that
+ // impossible by construction rather than by argument (a transferring
+ // replica withholds acks, so it should not be committing, but "should"
+ // is not an invariant this path can rely on).
+ //
+ // Deadlock-free: nothing between here and the write awaits, and
+ // `write_superblock` takes no lock of its own -- the checkpoint path
+ // calls it under this same gate.
+ let _superblock_gate = if snapshot_ahead && self.superblock.is_some() {
Review Comment:
the "impossible by construction" claim doesn't hold: `checkpoint_if_needed` holds `checkpoint_lock` for its whole sequence and takes `superblock_lock` only around the pairing write, while `persist_snapshot`'s snapshot.bin rewrite runs under `checkpoint_lock` alone - so this gate serializes nothing against the checkpoint's file rewrite. checkpoints run on spawned tasks, so a prepare that passed preflight before the transfer armed can drive one during this install's superblock await; the reachable damage is a `CheckpointChecksumMismatch` boot refusal if the node crashes inside that window (same op, different bytes). taking `checkpoint_lock` then `superblock_lock` here - the checkpoint's own order, so no inversion - makes the comment true.
##########
core/metadata/src/stm/mux.rs:
##########
@@ -188,6 +188,26 @@ where
}
}
+impl<SnapshotData, Head, Tail> RestoreSnapshotInPlace<SnapshotData> for variadic!(Head, ...Tail)
+where
+ Head: RestoreSnapshotInPlace<SnapshotData>,
+ Tail: RestoreSnapshotInPlace<SnapshotData>,
+{
+ fn restore_snapshot_in_place(&self, snapshot: &SnapshotData) -> Result<(), SnapshotError> {
+ self.0.restore_snapshot_in_place(snapshot)?;
Review Comment:
the two restores run sequentially with `?` and no rollback: if users restores and streams fails, the halves are left at different states, the snapshot was already persisted and the pairing seeded, and the caller logs and keeps running - the follow-up `commit_journal` then replays whatever contiguous suffix the local WAL holds into the half that did restore. the same malformed input fail-stops the boot path (`restore_snapshot?` aborts recovery). validating all sections are present before mutating anything - or fail-stopping on a partial restore - restores the symmetry.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -1296,6 +1336,24 @@ where
}
}
+/// One state-transfer serving payload.
+///
+/// The on-disk snapshot bytes plus the live client table, both
+/// frontier-stamped. Built by [`IggyMetadata::state_transfer_offer`] on the
+/// serving primary; the shard caches it per requester and serves chunks
+/// from it.
+pub struct StateTransferOffer {
+ /// Serving primary's applied frontier when the offer was built; the
+ /// receiver's tail repair targets past this.
+ pub commit_op: u64,
+ /// Manifest entries, index-aligned with `payloads`. Metadata plane:
+ /// `[METADATA_SNAPSHOT (frontier = sequence_number),
+ /// CLIENT_TABLE (frontier = commit_min at encode)]`.
+ pub artifacts: Vec<consensus::StateArtifact>,
+ /// Artifact bytes, chunk-served by `(manifest index, offset)`.
+ pub payloads: Vec<Vec<u8>>,
Review Comment:
`artifacts` and `payloads` are index-aligned by comment only, and the two halves are consumed in different files (manifest encoded from one, chunks served from the other). `Vec<(StateArtifact, Vec<u8>)>` makes the desync unrepresentable.
##########
core/shard/src/lib.rs:
##########
@@ -2749,6 +2964,790 @@ where
.await;
}
+ /// Start metadata tail journal-repair from `peer` when the commit walk
+ /// gap-stopped below the known frontier. Shared by `StartView` adoption
+ /// and the post-install step of a state transfer.
+ #[allow(clippy::future_not_send)]
+ async fn maybe_request_metadata_repair<P>(&self, consensus: &VsrConsensus<B, P>, peer: u8)
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ if consensus.is_normal()
+ && !consensus.is_transferring()
+ && consensus.commit_min() < consensus.commit_max()
+ && self.metadata_repair.borrow().is_none()
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ let to_op = consensus.commit_max();
+ let from_op = consensus.commit_min() + 1;
+ *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
+ nonce,
+ to_op,
+ peer,
+ idle_ticks: 0,
+ });
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ "metadata behind the group frontier; requesting repair"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
+
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn send_request_state_transfer<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ target: u8,
+ nonce: u128,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let msg =
+ Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
+ .transmute_header(|_, h: &mut RequestStateTransferHeader| {
+ h.command = Command2::RequestStateTransfer;
+ h.cluster = consensus.cluster();
+ h.replica = consensus.replica();
+ h.nonce = nonce;
+ h.namespace = consensus.namespace();
+ h.size = size_of::<RequestStateTransferHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
+ /// `available = 0` (the requester falls back to journal repair or
+ /// retries elsewhere); an offer ships its encoded state manifest as the
+ /// frame body.
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_state_transfer_target(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ offer: Option<&metadata::StateTransferOffer>,
+ ) where
+ B: MessageBus,
+ {
+ let manifest = offer.map(|offer| consensus::encode_state_manifest(&offer.artifacts));
+ let total_size =
+ size_of::<StateTransferTargetHeader>() + manifest.as_ref().map_or(0, Vec::len);
+ let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
+ if let Some(manifest) = &manifest {
+ msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ }
+ let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
+ h.command = Command2::StateTransferTarget;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.size = total_size as u32;
+ if let Some(offer) = offer {
+ h.available = 1;
+ h.commit_op = offer.commit_op;
+ }
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_request_state_chunk(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ artifact: u32,
+ offset: u64,
+ len: u32,
+ ) where
+ B: MessageBus,
+ {
+ let msg = Message::<RequestStateChunkHeader>::new(size_of::<RequestStateChunkHeader>())
+ .transmute_header(|_, h: &mut RequestStateChunkHeader| {
+ h.command = Command2::RequestStateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.artifact = artifact;
+ h.offset = offset;
+ h.len = len;
+ h.size = size_of::<RequestStateChunkHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Serve one `RequestStateTransfer`: build a fresh offer (or refuse),
+ /// cache it for the chunk pulls, and answer with the descriptor.
+ #[allow(clippy::future_not_send)]
+ async fn on_request_state_transfer(&self, msg: &Message<RequestStateTransferHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let offer = planes.0.state_transfer_offer();
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+ if let Some(offer) = offer {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ artifacts = offer.artifacts.len(),
+ total_len = offer.artifacts.iter().map(|a| a.len).sum::<u64>(),
+ "serving metadata state transfer"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ Some(&offer),
+ )
+ .await;
+ self.metadata_transfer_offers.borrow_mut().insert(
+ header.replica,
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer,
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ } else {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "cannot serve metadata state transfer (not a caught-up \
+ primary, or no snapshot persisted); requester falls back"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ }
+
+ /// Receiver side of the descriptor: accept it and start pulling chunks,
+ /// or fall back to journal repair when the peer cannot serve.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_transfer_target(&self, msg: &Message<StateTransferTargetHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ /// Alloc cap per artifact: a corrupt length field must not OOM the
+ /// shard. Far above any real metadata snapshot or client table.
+ const ARTIFACT_LEN_MAX: u64 = 1 << 30;
+
+ /// Alloc cap across the WHOLE manifest. The per-artifact cap alone does
+ /// not bound the total: `STATE_MANIFEST_ENTRIES_MAX` allows 65k entries,
+ /// and the buffers below are reserved eagerly, so per-artifact limits
+ /// would still admit a 64 TiB reservation and abort the process. The
+ /// manifest checksum only proves it arrived intact, not that the peer
+ /// computed it sanely, so bound what this side is willing to reserve.
+ const MANIFEST_TOTAL_LEN_MAX: u64 = 4 << 30;
+
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let session_matches = self
+ .metadata_transfer
+ .borrow()
+ .as_ref()
+ .is_some_and(|session| session.nonce == header.nonce);
+ if !session_matches {
+ return;
+ }
+
+ if header.available == 0 {
+ // The peer cannot serve. If we have never installed anything the
+ // local recovery stands; run the deferred commit walk and let
+ // journal repair cover the gap (a peer that never checkpointed
+ // retains its full WAL, so repair CAN cover it).
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ "state transfer unavailable; falling back to journal repair"
+ );
+ *self.metadata_transfer.borrow_mut() = None;
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, header.replica)
+ .await;
+ return;
+ }
+
+ // The manifest rides the body; a well-formed available=1 descriptor
+ // always carries one (an empty manifest still encodes its envelope).
+ let manifest = match consensus::decode_state_manifest(
+ &msg.as_slice()[size_of::<StateTransferTargetHeader>()..header.size as usize],
+ ) {
+ Ok(manifest) => manifest,
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ %error,
+ "state transfer descriptor manifest undecodable; ignoring"
+ );
+ return;
+ }
+ };
+ if let Some(oversized) = manifest.iter().find(|entry| entry.len > ARTIFACT_LEN_MAX) {
+ tracing::error!(
+ shard = self.id,
+ kind = oversized.kind,
+ len = oversized.len,
+ "state transfer descriptor exceeds artifact cap; ignoring"
+ );
+ return;
+ }
+ let declared_total = manifest
+ .iter()
+ .fold(0u64, |total, entry| total.saturating_add(entry.len));
+ if declared_total > MANIFEST_TOTAL_LEN_MAX {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ declared_total,
+ "state transfer descriptor exceeds the total manifest cap; ignoring"
+ );
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.target_accepted {
+ // Duplicate descriptor (stall retry crossed the original).
+ return;
+ }
+ session.target_accepted = true;
+ session.commit_op = header.commit_op;
+ // Under ARTIFACT_LEN_MAX (checked above), so the casts hold.
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ session.artifacts = manifest
+ .iter()
+ .map(|&entry| ArtifactProgress {
+ entry,
+ buf: Vec::with_capacity(entry.len as usize),
+ })
+ .collect();
+ }
+ session.idle_ticks = 0;
+ }
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Fetching);
+ }
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ total_len = manifest.iter().map(|entry| entry.len).sum::<u64>(),
+ commit_op = header.commit_op,
+ "state transfer target accepted; fetching"
+ );
+ self.on_transfer_progress().await;
+ }
+
+ /// Ask for the next missing chunk of the in-flight transfer (artifacts
+ /// pulled in manifest order). No-op when nothing is missing or no
+ /// manifest is accepted yet; also the stall-retry re-request.
+ #[allow(clippy::future_not_send)]
+ async fn request_pending_state_chunk(&self)
+ where
+ B: MessageBus,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let request = {
+ let session = self.metadata_transfer.borrow();
+ session.as_ref().and_then(|session| {
+ if !session.target_accepted {
+ return None;
+ }
+ let (index, artifact) = session
+ .artifacts
+ .iter()
+ .enumerate()
+ .find(|(_, artifact)| !artifact.complete())?;
+ let offset = artifact.buf.len() as u64;
+ let remaining = artifact.entry.len - offset;
+ #[allow(clippy::cast_possible_truncation)]
+ let len = remaining.min(u64::from(STATE_CHUNK_LEN)) as u32;
+ #[allow(clippy::cast_possible_truncation)]
+ Some((session.nonce, session.peer, index as u32, offset, len))
+ })
+ };
+ if let Some((nonce, peer, artifact, offset, len)) = request {
+ self.send_request_state_chunk(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ consensus.namespace(),
+ artifact,
+ offset,
+ len,
+ )
+ .await;
+ }
+ }
+
+ /// Serve one chunk out of the cached offer. An unknown nonce (offer
+ /// evicted, e.g. the serving process restarted) answers with an
+ /// `available = 0` descriptor so the requester restarts its session.
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn on_request_state_chunk(&self, msg: &Message<RequestStateChunkHeader>)
+ where
+ B: MessageBus,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+
+ // Frame built inside the borrow; every send runs after it drops (a
+ // RefCell borrow must not cross an await on the shard).
+ // Out-of-bounds requests are dropped silently inside the block.
+ let reply = {
+ let mut offers = self.metadata_transfer_offers.borrow_mut();
+ let served = offers
+ .get_mut(&header.replica)
+ .filter(|served| served.nonce == header.nonce);
+ served.map_or(Some(ChunkReply::UnknownOffer), |served| {
+ // Serving a chunk is the only liveness signal the offer gets;
+ // the expiry sweep drops it once these stop arriving.
+ served.idle_ticks = 0;
+ // Manifest-index addressing: an index past the offer is a
+ // requester bug (or a stale frame) and is dropped below.
+ let last_artifact = served.offer.payloads.len().saturating_sub(1);
+ let artifact_bytes = served.offer.payloads.get(header.artifact as usize)?;
+ let start = header.offset as usize;
+ let end = start.saturating_add(header.len as usize);
Review Comment:
`header.len` comes off the wire and only `len == 0` is rejected, so a peer asking for `u32::MAX` gets one frame of the whole remaining artifact. anything above the configured `message_bus.max_message_size` is rejected by the receiving transport, which tears down the entire replica connection - logged at `debug!` only, so at default level the operator just sees a link flap with no reason. the reverse config also breaks: `max_message_size` below `STATE_CHUNK_LEN` + header makes any artifact bigger than the cap untransferable (the const's doc says the bus cap is "far above this", but nothing checks it). clamp the served len to `min(STATE_CHUNK_LEN, max_message_size - HEADER_SIZE)`, derive the requester's chunk size from the same value, and validate the relation at boot.
##########
core/shard/src/lib.rs:
##########
@@ -2749,6 +2964,790 @@ where
.await;
}
+ /// Start metadata tail journal-repair from `peer` when the commit walk
+ /// gap-stopped below the known frontier. Shared by `StartView` adoption
+ /// and the post-install step of a state transfer.
+ #[allow(clippy::future_not_send)]
+ async fn maybe_request_metadata_repair<P>(&self, consensus: &VsrConsensus<B, P>, peer: u8)
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ if consensus.is_normal()
+ && !consensus.is_transferring()
+ && consensus.commit_min() < consensus.commit_max()
+ && self.metadata_repair.borrow().is_none()
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ let to_op = consensus.commit_max();
+ let from_op = consensus.commit_min() + 1;
+ *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
+ nonce,
+ to_op,
+ peer,
+ idle_ticks: 0,
+ });
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ "metadata behind the group frontier; requesting repair"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
+
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn send_request_state_transfer<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ target: u8,
+ nonce: u128,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let msg =
+ Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
+ .transmute_header(|_, h: &mut RequestStateTransferHeader| {
+ h.command = Command2::RequestStateTransfer;
+ h.cluster = consensus.cluster();
+ h.replica = consensus.replica();
+ h.nonce = nonce;
+ h.namespace = consensus.namespace();
+ h.size = size_of::<RequestStateTransferHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
+ /// `available = 0` (the requester falls back to journal repair or
+ /// retries elsewhere); an offer ships its encoded state manifest as the
+ /// frame body.
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_state_transfer_target(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ offer: Option<&metadata::StateTransferOffer>,
+ ) where
+ B: MessageBus,
+ {
+ let manifest = offer.map(|offer| consensus::encode_state_manifest(&offer.artifacts));
+ let total_size =
+ size_of::<StateTransferTargetHeader>() + manifest.as_ref().map_or(0, Vec::len);
+ let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
+ if let Some(manifest) = &manifest {
+ msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ }
+ let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
+ h.command = Command2::StateTransferTarget;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.size = total_size as u32;
+ if let Some(offer) = offer {
+ h.available = 1;
+ h.commit_op = offer.commit_op;
+ }
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_request_state_chunk(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ artifact: u32,
+ offset: u64,
+ len: u32,
+ ) where
+ B: MessageBus,
+ {
+ let msg = Message::<RequestStateChunkHeader>::new(size_of::<RequestStateChunkHeader>())
+ .transmute_header(|_, h: &mut RequestStateChunkHeader| {
+ h.command = Command2::RequestStateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.artifact = artifact;
+ h.offset = offset;
+ h.len = len;
+ h.size = size_of::<RequestStateChunkHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Serve one `RequestStateTransfer`: build a fresh offer (or refuse),
+ /// cache it for the chunk pulls, and answer with the descriptor.
+ #[allow(clippy::future_not_send)]
+ async fn on_request_state_transfer(&self, msg: &Message<RequestStateTransferHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let offer = planes.0.state_transfer_offer();
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+ if let Some(offer) = offer {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ artifacts = offer.artifacts.len(),
+ total_len = offer.artifacts.iter().map(|a| a.len).sum::<u64>(),
+ "serving metadata state transfer"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ Some(&offer),
+ )
+ .await;
+ self.metadata_transfer_offers.borrow_mut().insert(
+ header.replica,
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer,
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ } else {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "cannot serve metadata state transfer (not a caught-up \
+ primary, or no snapshot persisted); requester falls back"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ }
+
+ /// Receiver side of the descriptor: accept it and start pulling chunks,
+ /// or fall back to journal repair when the peer cannot serve.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_transfer_target(&self, msg: &Message<StateTransferTargetHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ /// Alloc cap per artifact: a corrupt length field must not OOM the
+ /// shard. Far above any real metadata snapshot or client table.
+ const ARTIFACT_LEN_MAX: u64 = 1 << 30;
+
+ /// Alloc cap across the WHOLE manifest. The per-artifact cap alone does
+ /// not bound the total: `STATE_MANIFEST_ENTRIES_MAX` allows 65k entries,
+ /// and the buffers below are reserved eagerly, so per-artifact limits
+ /// would still admit a 64 TiB reservation and abort the process. The
+ /// manifest checksum only proves it arrived intact, not that the peer
+ /// computed it sanely, so bound what this side is willing to reserve.
+ const MANIFEST_TOTAL_LEN_MAX: u64 = 4 << 30;
+
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let session_matches = self
+ .metadata_transfer
+ .borrow()
+ .as_ref()
+ .is_some_and(|session| session.nonce == header.nonce);
+ if !session_matches {
+ return;
+ }
+
+ if header.available == 0 {
+ // The peer cannot serve. If we have never installed anything the
+ // local recovery stands; run the deferred commit walk and let
+ // journal repair cover the gap (a peer that never checkpointed
+ // retains its full WAL, so repair CAN cover it).
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ "state transfer unavailable; falling back to journal repair"
+ );
+ *self.metadata_transfer.borrow_mut() = None;
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, header.replica)
+ .await;
+ return;
+ }
+
+ // The manifest rides the body; a well-formed available=1 descriptor
+ // always carries one (an empty manifest still encodes its envelope).
+ let manifest = match consensus::decode_state_manifest(
+ &msg.as_slice()[size_of::<StateTransferTargetHeader>()..header.size as usize],
+ ) {
+ Ok(manifest) => manifest,
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ %error,
+ "state transfer descriptor manifest undecodable; ignoring"
+ );
+ return;
+ }
+ };
+ if let Some(oversized) = manifest.iter().find(|entry| entry.len > ARTIFACT_LEN_MAX) {
+ tracing::error!(
+ shard = self.id,
+ kind = oversized.kind,
+ len = oversized.len,
+ "state transfer descriptor exceeds artifact cap; ignoring"
+ );
+ return;
+ }
+ let declared_total = manifest
+ .iter()
+ .fold(0u64, |total, entry| total.saturating_add(entry.len));
+ if declared_total > MANIFEST_TOTAL_LEN_MAX {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ declared_total,
+ "state transfer descriptor exceeds the total manifest cap; ignoring"
+ );
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.target_accepted {
+ // Duplicate descriptor (stall retry crossed the original).
+ return;
+ }
+ session.target_accepted = true;
+ session.commit_op = header.commit_op;
+ // Under ARTIFACT_LEN_MAX (checked above), so the casts hold.
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ session.artifacts = manifest
+ .iter()
+ .map(|&entry| ArtifactProgress {
+ entry,
+ buf: Vec::with_capacity(entry.len as usize),
+ })
+ .collect();
+ }
+ session.idle_ticks = 0;
+ }
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Fetching);
+ }
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ total_len = manifest.iter().map(|entry| entry.len).sum::<u64>(),
+ commit_op = header.commit_op,
+ "state transfer target accepted; fetching"
+ );
+ self.on_transfer_progress().await;
+ }
+
+ /// Ask for the next missing chunk of the in-flight transfer (artifacts
+ /// pulled in manifest order). No-op when nothing is missing or no
+ /// manifest is accepted yet; also the stall-retry re-request.
+ #[allow(clippy::future_not_send)]
+ async fn request_pending_state_chunk(&self)
+ where
+ B: MessageBus,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let request = {
+ let session = self.metadata_transfer.borrow();
+ session.as_ref().and_then(|session| {
+ if !session.target_accepted {
+ return None;
+ }
+ let (index, artifact) = session
+ .artifacts
+ .iter()
+ .enumerate()
+ .find(|(_, artifact)| !artifact.complete())?;
+ let offset = artifact.buf.len() as u64;
+ let remaining = artifact.entry.len - offset;
+ #[allow(clippy::cast_possible_truncation)]
+ let len = remaining.min(u64::from(STATE_CHUNK_LEN)) as u32;
+ #[allow(clippy::cast_possible_truncation)]
+ Some((session.nonce, session.peer, index as u32, offset, len))
+ })
+ };
+ if let Some((nonce, peer, artifact, offset, len)) = request {
+ self.send_request_state_chunk(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ consensus.namespace(),
+ artifact,
+ offset,
+ len,
+ )
+ .await;
+ }
+ }
+
+ /// Serve one chunk out of the cached offer. An unknown nonce (offer
+ /// evicted, e.g. the serving process restarted) answers with an
+ /// `available = 0` descriptor so the requester restarts its session.
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn on_request_state_chunk(&self, msg: &Message<RequestStateChunkHeader>)
+ where
+ B: MessageBus,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+
+ // Frame built inside the borrow; every send runs after it drops (a
+ // RefCell borrow must not cross an await on the shard).
+ // Out-of-bounds requests are dropped silently inside the block.
+ let reply = {
+ let mut offers = self.metadata_transfer_offers.borrow_mut();
+ let served = offers
+ .get_mut(&header.replica)
+ .filter(|served| served.nonce == header.nonce);
+ served.map_or(Some(ChunkReply::UnknownOffer), |served| {
+ // Serving a chunk is the only liveness signal the offer gets;
+ // the expiry sweep drops it once these stop arriving.
+ served.idle_ticks = 0;
+ // Manifest-index addressing: an index past the offer is a
+ // requester bug (or a stale frame) and is dropped below.
+ let last_artifact = served.offer.payloads.len().saturating_sub(1);
+ let artifact_bytes = served.offer.payloads.get(header.artifact as usize)?;
+ let start = header.offset as usize;
+ let end = start.saturating_add(header.len as usize);
+ // Tail of the final artifact: the receiver now holds everything
+ // the manifest promised, so the offer only has to outlive a
+ // possible re-request of this very chunk.
+ if header.artifact as usize == last_artifact && end >= artifact_bytes.len() {
+ served.fully_served = true;
+ }
+ artifact_bytes
+ .get(start..end.min(artifact_bytes.len()))
+ .map(|payload| {
+ let total_size = size_of::<StateChunkHeader>() + payload.len();
+ let mut chunk = Message::<StateChunkHeader>::new(total_size);
+ chunk.as_mut_slice()[size_of::<StateChunkHeader>()..]
+ .copy_from_slice(payload);
+ ChunkReply::Chunk(chunk.transmute_header(|_, h: &mut StateChunkHeader| {
+ h.command = Command2::StateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = header.nonce;
+ h.namespace = header.namespace;
+ h.artifact = header.artifact;
+ h.offset = header.offset;
+ h.size = total_size as u32;
+ }))
+ })
+ })
+ };
+ match reply {
+ Some(ChunkReply::Chunk(chunk)) => {
+ let _ = self
+ .bus
+ .send_to_replica(header.replica, chunk.into_generic().into_frozen())
+ .await;
+ }
+ Some(ChunkReply::UnknownOffer) => {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "state chunk request for an unknown offer; telling requester to restart"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ None => {
+ tracing::warn!(
+ shard = self.id,
+ requester = header.replica,
+ artifact = header.artifact,
+ offset = header.offset,
+ "state chunk request out of artifact bounds; ignoring"
+ );
+ }
+ }
+ }
+
+ /// Receive one chunk; on the last one, verify + install + hand the tail
+ /// to journal repair.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_chunk(&self, msg: &Message<StateChunkHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.nonce != header.nonce || !session.target_accepted {
+ return;
+ }
+ let Some(artifact) = session.artifacts.get_mut(header.artifact as usize) else {
+ return;
+ };
+ let payload = &msg.as_slice()[size_of::<StateChunkHeader>()..header.size as usize];
+ // Chunks are pulled sequentially with one in flight; anything
+ // else is a duplicate or reorder and is dropped (the stall retry
+ // re-requests from the current frontier).
+ if header.offset != artifact.buf.len() as u64 {
+ return;
+ }
+ if artifact.buf.len() as u64 + payload.len() as u64 > artifact.entry.len {
+ tracing::warn!(
+ shard = self.id,
+ artifact = header.artifact,
+ "state chunk overruns the declared artifact length; dropping frame"
+ );
+ return;
+ }
+ artifact.buf.extend_from_slice(payload);
+ session.idle_ticks = 0;
+ }
+ self.on_transfer_progress().await;
+ }
+
+ /// Drive the in-flight transfer forward: request the next missing chunk,
+ /// or - once every artifact is complete - verify, decode, and install.
+ /// Shared by descriptor acceptance and chunk arrival, so a manifest whose
+ /// artifacts are already complete (all empty) installs without waiting
+ /// for a chunk that will never come.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_transfer_progress(&self)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let complete = {
+ let session = self.metadata_transfer.borrow();
+ match session.as_ref() {
+ Some(session) if session.target_accepted => {
+ session.artifacts.iter().all(ArtifactProgress::complete)
+ }
+ _ => return,
+ }
+ };
+ if !complete {
+ self.request_pending_state_chunk().await;
+ return;
+ }
+
+ // All bytes in: verify, decode, install.
+ let session = self
+ .metadata_transfer
+ .borrow_mut()
+ .take()
+ .expect("session checked above");
+ let peer = session.peer;
+ let commit_op = session.commit_op;
+ let attempts = session.attempts;
+
+ // Per-artifact integrity, then pick the pieces this plane installs.
+ // Unknown kinds are refused rather than skipped: an artifact the
+ // serving peer thought worth shipping but this receiver cannot
+ // install would otherwise be silently dropped.
+ let mut snapshot: Option<Vec<u8>> = None;
+ let mut table: Option<(Vec<u8>, u64)> = None;
+ let mut damaged = false;
+ for (index, artifact) in session.artifacts.into_iter().enumerate() {
+ let actual = consensus::state_artifact_checksum(&artifact.buf);
+ if actual != artifact.entry.checksum {
+ tracing::error!(
+ shard = self.id,
+ artifact = index,
+ kind = artifact.entry.kind,
+ "state transfer artifact checksum mismatch"
+ );
+ damaged = true;
+ break;
+ }
+ match artifact.entry.kind {
+ consensus::artifact_kind::METADATA_SNAPSHOT => snapshot = Some(artifact.buf),
+ consensus::artifact_kind::CLIENT_TABLE => {
+ table = Some((artifact.buf, artifact.entry.frontier));
+ }
+ kind => {
+ tracing::error!(
+ shard = self.id,
+ kind,
+ "state transfer manifest carries a kind this plane cannot install"
+ );
+ damaged = true;
+ break;
+ }
+ }
+ }
+
+ let decoded = if damaged {
+ None
+ } else if let (Some(snapshot), Some((table_bytes, table_frontier))) = (snapshot, table) {
+ match consensus::ClientTable::decode(&table_bytes, self.clients_table_max.get()) {
+ Ok(table) => Some((snapshot, table, table_frontier)),
+ Err(error) => {
+ tracing::error!(shard = self.id, %error, "transferred client table undecodable");
+ None
+ }
+ }
+ } else {
+ tracing::error!(
+ shard = self.id,
+ "state transfer manifest is missing the snapshot or client table artifact"
+ );
+ None
+ };
+
+ let Some((snapshot, table, table_frontier)) = decoded else {
+ // Damage is usually transit corruption, which a re-fetch fixes. But
+ // it can also be permanent -- a peer whose artifacts this build
+ // cannot decode, or an unknown artifact kind -- and that re-offers
+ // identically every round. Carry the attempt count across the
+ // restart and share the stall budget, so a re-fetch that keeps
+ // failing gives up instead of pulling the whole snapshot forever.
+ // Distinct from the stall path: frames ARE flowing here, so
+ // `idle_ticks` never accumulates and that sweep can never fire.
+ let attempts = attempts + 1;
+ if attempts > STATE_TRANSFER_MAX_STALL_RETRIES {
+ tracing::warn!(
+ shard = self.id,
+ peer,
+ attempts,
+ "state transfer artifacts kept failing to decode; abandoning \
+ and falling back to journal repair"
+ );
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, peer).await;
+ return;
+ }
+ // Restart the session from scratch against the same peer (fresh
+ // nonce; the peer re-offers).
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::Fetching {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::AwaitingTarget);
+ }
+ let nonce = iggy_common::random_id::get_uuid();
+ *self.metadata_transfer.borrow_mut() = Some(MetadataTransferSession {
+ nonce,
+ peer,
+ commit_op: 0,
+ artifacts: Vec::new(),
+ target_accepted: false,
+ attempts,
+ idle_ticks: 0,
+ });
+ self.send_request_state_transfer(consensus, peer, nonce)
+ .await;
+ return;
+ };
+
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Installing);
+ match planes
+ .0
+ .install_state_transfer(&snapshot, table, table_frontier, commit_op)
+ .await
+ {
+ Ok(applied_frontier) => {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ // `applied_frontier`, not the transferred snapshot's op: the install
+ // returns `max(snapshot_seq, local_applied)`, which differs whenever a
+ // serving peer offers a snapshot BEHIND this replica (checkpoints are
+ // node-local) and the local state machine is kept instead.
+ tracing::info!(
+ shard = self.id,
+ applied_frontier,
+ commit_op,
+ table_frontier,
+ "metadata state transfer installed; handing tail to journal repair"
+ );
+ // Walk whatever is already walkable, then let repair fetch
+ // the (snapshot_seq, commit_max] tail.
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, peer).await;
+ }
+ // Installed, but its `(checkpoint_op, checksum)` pairing is not durable
+ // yet. Not a failed install: the snapshot, table and frontiers are all in
+ // place, and the coordinator already holds the new pairing, so the next
+ // superblock write (view change or checkpoint) records it. Until then a
+ // crash recovers the PREVIOUS checkpoint and this replica transfers again,
+ // which is correct, just wasted work. Continue as a success.
+ Err(metadata::stm::snapshot::SnapshotError::SuperblockNotDurable { op }) => {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ tracing::warn!(
+ shard = self.id,
+ snapshot_seq = op,
+ commit_op,
+ "metadata state transfer installed but its checkpoint pairing is \
Review Comment:
two related papercuts. `SuperblockNotDurable` is an `Err` for an outcome that is a success - the install fully landed (snapshot, table, frontiers, commit point) before it's returned, and this sole caller special-cases it back; any future caller treating err as failure would lose a completed install. and this degraded log shares the prefix "metadata state transfer installed" with the success line while `stdout_occurrences` matches substrings, so 6 of the 7 new specs pass on the degraded path (only `assert_pairing_matches_install` still discriminates). `Result<InstallOutcome, SnapshotError>` plus distinct log strings fixes both.
##########
core/shard/src/lib.rs:
##########
@@ -1440,7 +1561,10 @@ where
Input = Message<PrepareHeader>,
Output = metadata::stm::result::ApplyReply,
Error = iggy_common::IggyError,
- > + StreamsFrontend,
+ > + StreamsFrontend
+ + metadata::stm::snapshot::RestoreSnapshotInPlace<
Review Comment:
this same 3-part bound (`StateMachine<...> + StreamsFrontend + RestoreSnapshotInPlace<MetadataSnapshot>`) is now pasted 11 times across this file and router.rs. a `trait MetadataStm: ... {}` with a blanket impl collapses all of them with no api change.
##########
core/consensus/src/client_table.rs:
##########
@@ -944,6 +952,200 @@ impl ClientTable {
}
}
+/// Failure decoding an encoded client table (state transfer install path).
+#[derive(Debug)]
+pub enum ClientTableCodecError {
+ /// Byte stream ended mid-field.
+ Truncated,
+ /// Leading magic is not [`CLIENT_TABLE_MAGIC`].
+ BadMagic,
+ /// Trailing hash does not match the content.
+ ChecksumMismatch { expected: u64, actual: u64 },
+ /// Encoded entry count exceeds this table's capacity.
+ TooManyEntries { count: u32, max: usize },
+ /// A cached reply's bytes do not parse as a valid reply message.
+ InvalidReply,
+ /// An entry carries an empty reply ring (violates the never-empty
+ /// invariant registration establishes).
+ EmptyRing,
+}
+
+impl std::fmt::Display for ClientTableCodecError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Truncated => write!(f, "encoded client table truncated"),
+ Self::BadMagic => write!(f, "encoded client table has wrong magic"),
+ Self::ChecksumMismatch { expected, actual } => write!(
+ f,
+ "client table checksum mismatch: expected {expected:#018x}, actual {actual:#018x}"
+ ),
+ Self::TooManyEntries { count, max } => {
+ write!(f, "encoded client table holds {count} entries, max {max}")
+ }
+ Self::InvalidReply => write!(f, "encoded client table holds an invalid cached reply"),
+ Self::EmptyRing => write!(f, "encoded client table entry has an empty reply ring"),
+ }
+ }
+}
+
+impl std::error::Error for ClientTableCodecError {}
+
+/// Format tag for [`ClientTable::encode`]; bump on layout change.
+pub const CLIENT_TABLE_MAGIC: [u8; 4] = *b"ICT1";
+
+impl ClientTable {
+ /// Encode the table for state transfer.
+ ///
+ /// Layout (all little-endian): `magic(4) count(u32)` then per entry in
+ /// slot order `client(u128) epoch(u64) user_id(u32) watermark(u64)
+ /// watermark_checksum(u128) ring_len(u8) [reply_len(u32) reply_bytes]*`,
+ /// terminated by an `XxHash3_64(8)` over everything before it.
+ ///
+ /// Slot order makes the bytes deterministic across caught-up replicas:
+ /// slots are apply-derived state.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn encode(&self) -> Vec<u8> {
+ let mut out = Vec::with_capacity(64 * self.index.len() + 16);
Review Comment:
`64 * self.index.len() + 16` under-reserves badly: the fixed fields alone are 53 bytes per entry and each cached reply is a full wire message (>= 256 bytes) plus a 4-byte length, ring up to 5. at the default 8192-client cap a full table encodes to roughly 11 MB grown from a half-MB reservation - several reallocs of a multi-MB buffer on the serving primary's pump, per offer build. summing the actual ring byte lengths first makes it one allocation.
##########
core/shard/src/lib.rs:
##########
@@ -2749,6 +2964,790 @@ where
.await;
}
+ /// Start metadata tail journal-repair from `peer` when the commit walk
+ /// gap-stopped below the known frontier. Shared by `StartView` adoption
+ /// and the post-install step of a state transfer.
+ #[allow(clippy::future_not_send)]
+ async fn maybe_request_metadata_repair<P>(&self, consensus: &VsrConsensus<B, P>, peer: u8)
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ if consensus.is_normal()
+ && !consensus.is_transferring()
+ && consensus.commit_min() < consensus.commit_max()
+ && self.metadata_repair.borrow().is_none()
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ let to_op = consensus.commit_max();
+ let from_op = consensus.commit_min() + 1;
+ *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
+ nonce,
+ to_op,
+ peer,
+ idle_ticks: 0,
+ });
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ "metadata behind the group frontier; requesting repair"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
+
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn send_request_state_transfer<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ target: u8,
+ nonce: u128,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let msg =
+ Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
+ .transmute_header(|_, h: &mut RequestStateTransferHeader| {
+ h.command = Command2::RequestStateTransfer;
+ h.cluster = consensus.cluster();
+ h.replica = consensus.replica();
+ h.nonce = nonce;
+ h.namespace = consensus.namespace();
+ h.size = size_of::<RequestStateTransferHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
+ /// `available = 0` (the requester falls back to journal repair or
+ /// retries elsewhere); an offer ships its encoded state manifest as the
+ /// frame body.
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_state_transfer_target(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ offer: Option<&metadata::StateTransferOffer>,
+ ) where
+ B: MessageBus,
+ {
+ let manifest = offer.map(|offer| consensus::encode_state_manifest(&offer.artifacts));
+ let total_size =
+ size_of::<StateTransferTargetHeader>() + manifest.as_ref().map_or(0, Vec::len);
+ let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
+ if let Some(manifest) = &manifest {
+ msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ }
+ let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
+ h.command = Command2::StateTransferTarget;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.size = total_size as u32;
+ if let Some(offer) = offer {
+ h.available = 1;
+ h.commit_op = offer.commit_op;
+ }
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ #[allow(
+ clippy::future_not_send,
+ clippy::cast_possible_truncation,
+ clippy::too_many_arguments
+ )]
+ async fn send_request_state_chunk(
+ &self,
+ cluster: u128,
+ self_id: u8,
+ target: u8,
+ nonce: u128,
+ namespace: u64,
+ artifact: u32,
+ offset: u64,
+ len: u32,
+ ) where
+ B: MessageBus,
+ {
+ let msg = Message::<RequestStateChunkHeader>::new(size_of::<RequestStateChunkHeader>())
+ .transmute_header(|_, h: &mut RequestStateChunkHeader| {
+ h.command = Command2::RequestStateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = nonce;
+ h.namespace = namespace;
+ h.artifact = artifact;
+ h.offset = offset;
+ h.len = len;
+ h.size = size_of::<RequestStateChunkHeader>() as u32;
+ });
+ let _ = self
+ .bus
+ .send_to_replica(target, msg.into_generic().into_frozen())
+ .await;
+ }
+
+ /// Serve one `RequestStateTransfer`: build a fresh offer (or refuse),
+ /// cache it for the chunk pulls, and answer with the descriptor.
+ #[allow(clippy::future_not_send)]
+ async fn on_request_state_transfer(&self, msg: &Message<RequestStateTransferHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let offer = planes.0.state_transfer_offer();
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+ if let Some(offer) = offer {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ artifacts = offer.artifacts.len(),
+ total_len = offer.artifacts.iter().map(|a| a.len).sum::<u64>(),
+ "serving metadata state transfer"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ Some(&offer),
+ )
+ .await;
+ self.metadata_transfer_offers.borrow_mut().insert(
+ header.replica,
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer,
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ } else {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ "cannot serve metadata state transfer (not a caught-up \
+ primary, or no snapshot persisted); requester falls back"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
+ }
+
+ /// Receiver side of the descriptor: accept it and start pulling chunks,
+ /// or fall back to journal repair when the peer cannot serve.
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ async fn on_state_transfer_target(&self, msg: &Message<StateTransferTargetHeader>)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + metadata::stm::snapshot::RestoreSnapshotInPlace<
+ metadata::stm::snapshot::MetadataSnapshot,
+ >,
+ {
+ /// Alloc cap per artifact: a corrupt length field must not OOM the
+ /// shard. Far above any real metadata snapshot or client table.
+ const ARTIFACT_LEN_MAX: u64 = 1 << 30;
+
+ /// Alloc cap across the WHOLE manifest. The per-artifact cap alone does
+ /// not bound the total: `STATE_MANIFEST_ENTRIES_MAX` allows 65k entries,
+ /// and the buffers below are reserved eagerly, so per-artifact limits
+ /// would still admit a 64 TiB reservation and abort the process. The
+ /// manifest checksum only proves it arrived intact, not that the peer
+ /// computed it sanely, so bound what this side is willing to reserve.
+ const MANIFEST_TOTAL_LEN_MAX: u64 = 4 << 30;
+
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let session_matches = self
+ .metadata_transfer
+ .borrow()
+ .as_ref()
+ .is_some_and(|session| session.nonce == header.nonce);
+ if !session_matches {
+ return;
+ }
+
+ if header.available == 0 {
+ // The peer cannot serve. If we have never installed anything the
+ // local recovery stands; run the deferred commit walk and let
+ // journal repair cover the gap (a peer that never checkpointed
+ // retains its full WAL, so repair CAN cover it).
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ "state transfer unavailable; falling back to journal repair"
+ );
+ *self.metadata_transfer.borrow_mut() = None;
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ planes.0.commit_journal().await;
+ self.maybe_request_metadata_repair(consensus, header.replica)
+ .await;
+ return;
+ }
+
+ // The manifest rides the body; a well-formed available=1 descriptor
+ // always carries one (an empty manifest still encodes its envelope).
+ let manifest = match consensus::decode_state_manifest(
+ &msg.as_slice()[size_of::<StateTransferTargetHeader>()..header.size as usize],
+ ) {
+ Ok(manifest) => manifest,
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ %error,
+ "state transfer descriptor manifest undecodable; ignoring"
+ );
+ return;
+ }
+ };
+ if let Some(oversized) = manifest.iter().find(|entry| entry.len > ARTIFACT_LEN_MAX) {
+ tracing::error!(
+ shard = self.id,
+ kind = oversized.kind,
+ len = oversized.len,
+ "state transfer descriptor exceeds artifact cap; ignoring"
+ );
+ return;
+ }
+ let declared_total = manifest
+ .iter()
+ .fold(0u64, |total, entry| total.saturating_add(entry.len));
+ if declared_total > MANIFEST_TOTAL_LEN_MAX {
+ tracing::error!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ declared_total,
+ "state transfer descriptor exceeds the total manifest cap; ignoring"
+ );
+ return;
+ }
+
+ {
+ let mut session = self.metadata_transfer.borrow_mut();
+ let Some(session) = session.as_mut() else {
+ return;
+ };
+ if session.target_accepted {
+ // Duplicate descriptor (stall retry crossed the original).
+ return;
+ }
+ session.target_accepted = true;
+ session.commit_op = header.commit_op;
+ // Under ARTIFACT_LEN_MAX (checked above), so the casts hold.
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ session.artifacts = manifest
+ .iter()
+ .map(|&entry| ArtifactProgress {
+ entry,
+ buf: Vec::with_capacity(entry.len as usize),
+ })
+ .collect();
+ }
+ session.idle_ticks = 0;
+ }
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Fetching);
+ }
+ tracing::info!(
+ shard = self.id,
+ peer = header.replica,
+ artifacts = manifest.len(),
+ total_len = manifest.iter().map(|entry| entry.len).sum::<u64>(),
+ commit_op = header.commit_op,
+ "state transfer target accepted; fetching"
+ );
+ self.on_transfer_progress().await;
+ }
+
+ /// Ask for the next missing chunk of the in-flight transfer (artifacts
+ /// pulled in manifest order). No-op when nothing is missing or no
+ /// manifest is accepted yet; also the stall-retry re-request.
+ #[allow(clippy::future_not_send)]
+ async fn request_pending_state_chunk(&self)
+ where
+ B: MessageBus,
+ {
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ let request = {
+ let session = self.metadata_transfer.borrow();
+ session.as_ref().and_then(|session| {
+ if !session.target_accepted {
+ return None;
+ }
+ let (index, artifact) = session
+ .artifacts
+ .iter()
+ .enumerate()
+ .find(|(_, artifact)| !artifact.complete())?;
+ let offset = artifact.buf.len() as u64;
+ let remaining = artifact.entry.len - offset;
+ #[allow(clippy::cast_possible_truncation)]
+ let len = remaining.min(u64::from(STATE_CHUNK_LEN)) as u32;
+ #[allow(clippy::cast_possible_truncation)]
+ Some((session.nonce, session.peer, index as u32, offset, len))
+ })
+ };
+ if let Some((nonce, peer, artifact, offset, len)) = request {
+ self.send_request_state_chunk(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ consensus.namespace(),
+ artifact,
+ offset,
+ len,
+ )
+ .await;
+ }
+ }
+
+ /// Serve one chunk out of the cached offer. An unknown nonce (offer
+ /// evicted, e.g. the serving process restarted) answers with an
+ /// `available = 0` descriptor so the requester restarts its session.
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn on_request_state_chunk(&self, msg: &Message<RequestStateChunkHeader>)
+ where
+ B: MessageBus,
+ {
+ let header = *msg.header();
+ let planes = self.plane.inner();
+ let Some(ref consensus) = planes.0.consensus else {
+ return;
+ };
+ if consensus.namespace() != header.namespace {
+ return;
+ }
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+
+ // Frame built inside the borrow; every send runs after it drops (a
+ // RefCell borrow must not cross an await on the shard).
+ // Out-of-bounds requests are dropped silently inside the block.
+ let reply = {
+ let mut offers = self.metadata_transfer_offers.borrow_mut();
+ let served = offers
+ .get_mut(&header.replica)
+ .filter(|served| served.nonce == header.nonce);
+ served.map_or(Some(ChunkReply::UnknownOffer), |served| {
+ // Serving a chunk is the only liveness signal the offer gets;
+ // the expiry sweep drops it once these stop arriving.
+ served.idle_ticks = 0;
+ // Manifest-index addressing: an index past the offer is a
+ // requester bug (or a stale frame) and is dropped below.
+ let last_artifact = served.offer.payloads.len().saturating_sub(1);
+ let artifact_bytes = served.offer.payloads.get(header.artifact as usize)?;
+ let start = header.offset as usize;
+ let end = start.saturating_add(header.len as usize);
+ // Tail of the final artifact: the receiver now holds everything
+ // the manifest promised, so the offer only has to outlive a
+ // possible re-request of this very chunk.
+ if header.artifact as usize == last_artifact && end >= artifact_bytes.len() {
Review Comment:
`fully_served` is set before the bounds check decides whether anything is actually served, so an out-of-bounds or stale frame that ships zero bytes still flips a live offer onto the short expiry. and the short expiry itself is one `repair_retry_ticks` - the same interval as the receiver's stall re-request, plus a network hop - so the re-request-the-last-chunk window this grace exists for is roughly a coin flip, and losing it costs a full re-pull (expired offer -> `UnknownOffer` -> session dropped with everything already downloaded). move the flag inside the map that produces the payload, and give it at least 2 retry intervals.
##########
core/integration/tests/cluster/metadata_state_transfer.rs:
##########
@@ -0,0 +1,419 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Spec test for metadata state transfer: a node that restarts into a live
+//! cluster replaces its snapshot-shaped state (metadata snapshot + client
+//! table) from the current primary instead of relying on its own WAL.
+//!
+//! The scenario forces the interesting case: enough committed metadata ops
+//! to trip a checkpoint on every node (`journal_slots` shrunk below), which
+//! drains the early ops -- including the client's register -- out of every
+//! WAL. A restarted node's local recovery can then neither replay those ops
+//! nor journal-repair them (the serving peers evicted them too); only the
+//! transferred snapshot + table carry that history.
+//!
+//! The transferred node is a follower afterwards, so its installed state has
+//! no client-visible surface to assert against (followers neither commit nor
+//! serve resume lookups). The install is pinned via its log marker; the
+//! functional assert (post-restart continuation commits cluster-wide) rides
+//! on top.
+
+#![cfg(feature = "vsr")]
+
+use super::client_table_restart::{
+ commit_request, create_stream_payload, register, resume_request, tcp_addr, tcp_addrs,
+};
+use integration::harness::TestHarness;
+use integration::iggy_harness;
+use std::time::Duration;
+use tokio::time::{Instant, sleep};
+
+/// Committed ops before the restart. `journal_slots = 256` with the built-in
+/// checkpoint margin (64) forces a checkpoint at ~192 committed ops, so this
+/// guarantees at least one checkpoint+drain on every node, pushing the
+/// register (op 1) below every snapshot floor.
+const OPS_BEFORE_RESTART: u64 = 220;
+
+/// How long the restarted follower gets to probe, fetch, and install.
+const TRANSFER_BUDGET: Duration = Duration::from_secs(30);
+
+const MARKER_POLL: Duration = Duration::from_millis(200);
+
+#[iggy_harness(cluster_nodes = 3, server(metadata.journal_slots = "256"))]
+async fn given_checkpointed_cluster_when_node_restarts_should_state_transfer_metadata(
+ harness: &mut TestHarness,
+) {
+ let addr = tcp_addr(harness);
+ let (mut stream, session) = register(addr).await;
+ for request in 1..=OPS_BEFORE_RESTART {
+ commit_request(
+ &mut stream,
+ session,
+ request,
+ &create_stream_payload(&format!("iggy-transfer-{request}")),
+ )
+ .await;
+ }
+ drop(stream);
+
+ harness.restart_server().await.unwrap();
+
+ // Functional: the session (registered below every snapshot floor by now)
+ // still continues cluster-wide.
+ let addrs = tcp_addrs(harness);
+ // A continuation is a fresh op, so there is no cached reply to compare.
+ resume_request(
+ &addrs,
+ session,
+ OPS_BEFORE_RESTART + 1,
+ &create_stream_payload("iggy-transfer-continuation"),
+ None,
+ )
+ .await;
+
+ // The restarted node itself: it must have entered state transfer,
+ // fetched the primary's snapshot + client table, and installed them.
+ // Its own WAL no longer holds the pre-checkpoint history, so nothing
+ // short of the transfer can restore that state on it.
+ let deadline = Instant::now() + TRANSFER_BUDGET;
+ loop {
+ if harness
+ .node(0)
+ .stdout_contains("metadata state transfer installed")
+ {
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "restarted node never completed the metadata state transfer \
+ within {TRANSFER_BUDGET:?}"
+ );
+ sleep(MARKER_POLL).await;
+ }
+}
+
+/// A node that joins a live, already-checkpointed cluster with NO local
+/// history must state-transfer, exactly like the restart-with-WAL case above.
+///
+/// This is the shape the user cares about beyond a plain restart: a fresh
+/// operator-provisioned replacement, or the third node of a cluster whose
+/// first two formed quorum and committed past a checkpoint before it arrived.
+/// It has no WAL to probe from, so the restart-arms-transfer boot path does
+/// not apply -- it joins as a plain view-0 backup, learns the frontier from
+/// the primary's heartbeats, discovers via journal repair that the gap sits
+/// below the retained floor, and converts THAT into a state transfer. The
+/// same `RangeEvicted`-to-transfer path serves the restart, the late join,
+/// and a partition heal; only the way each reaches repair differs.
+///
+/// Node 2 (a follower) is wiped rather than node 0 so the client keeps its
+/// primary and the cluster never loses quorum -- a genuine late join, not a
+/// restart of the whole cluster.
+#[iggy_harness(cluster_nodes = 3, server(metadata.journal_slots = "256"))]
+async fn given_checkpointed_cluster_when_fresh_node_joins_late_should_state_transfer_metadata(
+ harness: &mut TestHarness,
+) {
+ let addr = tcp_addr(harness);
+ let (mut stream, session) = register(addr).await;
+ for request in 1..=OPS_BEFORE_RESTART {
+ commit_request(
+ &mut stream,
+ session,
+ request,
+ &create_stream_payload(&format!("iggy-latejoin-{request}")),
+ )
+ .await;
+ }
+
+ // Wipe a follower and bring it back with an empty data directory while the
+ // other two keep quorum. Its missing prefix (register at op 1 included) was
+ // checkpointed away on every survivor, so nothing but a transfer can seed
+ // its metadata state and client table.
+ harness.restart_node_from_clean_slate(2).unwrap();
+
+ let deadline = Instant::now() + TRANSFER_BUDGET;
+ loop {
+ if harness
+ .node(2)
+ .stdout_contains("metadata state transfer installed")
+ {
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "fresh late-joining node never completed the metadata state transfer \
+ within {TRANSFER_BUDGET:?}; it must convert a repair-floor eviction \
+ into a transfer, not wedge below the floor"
+ );
+ sleep(MARKER_POLL).await;
+ }
+
+ // Functional: the cluster still commits with the rejoined node present, and
+ // the original session (registered below every floor) continues.
+ drop(stream);
+ let addrs = tcp_addrs(harness);
+ resume_request(
+ &addrs,
+ session,
+ OPS_BEFORE_RESTART + 1,
+ &create_stream_payload("iggy-latejoin-continuation"),
+ None,
+ )
+ .await;
+
+ // The install replaced `snapshot.bin` with the primary's copy, so the
+ // superblock's `(checkpoint_op, checksum)` pairing has to name THAT snapshot.
+ // A stale pairing does not refuse boot -- `verify_checkpoint_pairing` reads
+ // `checkpoint_op < snapshot_op` as a lagging local checkpoint and accepts --
+ // which is exactly why this reads the durable record instead of inferring
+ // health from a successful restart: a pairing that names a snapshot no longer
+ // on disk silently disarms the integrity check it exists to power.
+ //
+ // A wiped node has `local_applied == 0`, so the transferred snapshot is always
+ // ahead and the pairing is always recorded; the marker carries the op it
+ // recorded, so the assertion compares against what install actually wrote.
+ assert_pairing_matches_install(harness, 2).await;
+}
+
+/// Assert `node`'s durable superblock pairs with the checkpoint its state-transfer
+/// install recorded.
+async fn assert_pairing_matches_install(harness: &TestHarness, node: usize) {
+ const PAIRING_MARKER: &str = "state transfer recorded its checkpoint pairing";
+
+ // The server colors its tracing fields, so `checkpoint_op=193` reaches the log
+ // as `checkpoint_op\x1b[0m\x1b[2m=\x1b[0m193`. Strip escapes before matching,
+ // the same reason `ServerHandle::stdout_occurrences` does.
+ let (stdout, _stderr) = harness.node(node).collect_logs();
Review Comment:
verbatim copy of the harness's private `strip_ansi`. this caller needs the stripped text (it parses `checkpoint_op=` out of it), so `stdout_occurrences` can't serve it - export `strip_ansi` or add a `ServerHandle::stdout_plain()` and delete the copy.
##########
core/consensus/src/state_manifest.rs:
##########
@@ -0,0 +1,341 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! State-transfer manifest: the artifact list a serving peer offers, carried
+//! in the `StateTransferTarget` body.
+//!
+//! `TigerBeetle` pulls state as content-addressed grid blocks, so a bare
+//! `(address, checksum)` names any piece of state and one repair protocol
+//! covers manifest, free set, client sessions, and LSM tables alike. Iggy's
+//! artifacts (metadata snapshot, client table, partition segments) are not
+//! content-addressed, so the manifest supplies the addressing instead: each
+//! entry names one artifact, `(manifest index, byte offset)` is the chunk
+//! address the `RequestStateChunk`/`StateChunk` frames pull by, and the
+//! per-artifact checksum is the integrity stamp (chunks carry none).
+//!
+//! Plane-agnostic by construction: the metadata plane ships two entries
+//! (snapshot + client table); the partition plane ships N (segment logs,
+//! consumer offsets, ...) through the same frames by appending [`kind`]
+//! values. Entries carry their own length on the wire (`entry_len`), so new
+//! per-entry fields can be appended without breaking old decoders.
+//!
+//! [`kind`]: StateArtifact::kind
+
+use std::hash::Hasher;
+use twox_hash::XxHash3_64;
+
+/// Artifact kinds. Wire-pinned: never reorder or reuse.
+pub mod artifact_kind {
+ /// Metadata plane: `snapshot.bin` bytes verbatim.
+ pub const METADATA_SNAPSHOT: u8 = 0;
+ /// Metadata plane: [`crate::ClientTable::encode`] bytes.
+ pub const CLIENT_TABLE: u8 = 1;
+ // Partition plane (reserved, not served yet):
+ // SEGMENT_LOG = 2, CONSUMER_OFFSETS = 3.
+}
+
+/// One artifact a serving peer offers: what it is, where its receiver-side
+/// watermark sits, and how to verify the assembled bytes.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct StateArtifact {
+ /// [`artifact_kind`] value. Unknown kinds decode fine (forward compat);
+ /// the receiving plane decides whether it can install them.
+ pub kind: u8,
+ /// Kind-specific watermark: the snapshot's `sequence_number`, the client
+ /// table's mutation frontier, a segment's base offset, ...
+ pub frontier: u64,
+ /// Artifact byte length; `(manifest index, offset in 0..len)` addresses
+ /// every chunk of it.
+ pub len: u64,
+ /// `XxHash3_64` over the artifact bytes (artifact-level integrity;
+ /// chunks carry none).
+ pub checksum: u64,
+}
+
+impl StateArtifact {
+ /// Entry for `bytes`, stamping length + checksum.
+ #[must_use]
+ pub fn for_bytes(kind: u8, frontier: u64, bytes: &[u8]) -> Self {
+ Self {
+ kind,
+ frontier,
+ len: bytes.len() as u64,
+ checksum: state_artifact_checksum(bytes),
+ }
+ }
+}
+
+/// Artifact-level integrity stamp (manifest `checksum` field).
+#[must_use]
+pub fn state_artifact_checksum(bytes: &[u8]) -> u64 {
+ let mut hasher = XxHash3_64::new();
+ hasher.write(bytes);
+ hasher.finish()
+}
+
+/// Failure decoding an encoded manifest.
+#[derive(Debug)]
+pub enum StateManifestError {
+ /// Byte stream ended mid-field.
+ Truncated,
+ /// Leading magic is not [`STATE_MANIFEST_MAGIC`].
+ BadMagic,
+ /// Trailing hash does not match the content.
+ ChecksumMismatch { expected: u64, actual: u64 },
+ /// Entry count exceeds [`STATE_MANIFEST_ENTRIES_MAX`].
+ TooManyEntries { count: u32 },
+ /// Declared per-entry length is below this version's field set (a
+ /// FUTURE-shrunk entry; longer entries are fine, the tail is skipped).
+ EntryTooShort { entry_len: u8 },
+}
+
+impl std::fmt::Display for StateManifestError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Truncated => write!(f, "state manifest truncated"),
+ Self::BadMagic => write!(f, "state manifest has wrong magic"),
+ Self::ChecksumMismatch { expected, actual } => write!(
+ f,
+ "state manifest checksum mismatch: expected {expected:#018x}, actual {actual:#018x}"
+ ),
+ Self::TooManyEntries { count } => {
+ write!(
+ f,
+ "state manifest holds {count} entries, max {STATE_MANIFEST_ENTRIES_MAX}"
+ )
+ }
+ Self::EntryTooShort { entry_len } => write!(
+ f,
+ "state manifest entry length {entry_len} below this version's {STATE_MANIFEST_ENTRY_LEN}"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for StateManifestError {}
+
+/// Format tag for [`encode_state_manifest`]; bump on incompatible change
+/// (appending entry fields is compatible, see `entry_len`).
+pub const STATE_MANIFEST_MAGIC: [u8; 4] = *b"ISM1";
+
+/// This version's entry size: `kind(1) frontier(8) len(8) checksum(8)`.
+pub const STATE_MANIFEST_ENTRY_LEN: u8 = 25;
+
+/// Sanity ceiling on entries per manifest. A partition rejoin lists at most
+/// its retained segments plus a handful of trailers; 65k is a corruption
+/// guard, not a target.
+pub const STATE_MANIFEST_ENTRIES_MAX: u32 = 1 << 16;
+
+/// Encode a manifest.
+///
+/// Layout (little-endian): `magic(4) count(u32) entry_len(u8)` then per
+/// entry `kind(u8) frontier(u64) len(u64) checksum(u64)`, terminated by an
+/// `XxHash3_64(8)` over everything before it.
+///
+/// # Panics
+/// If `artifacts.len()` exceeds [`STATE_MANIFEST_ENTRIES_MAX`] (a serving
+/// peer offering that many artifacts is a bug, not an input).
+#[must_use]
+#[allow(clippy::cast_possible_truncation)]
+pub fn encode_state_manifest(artifacts: &[StateArtifact]) -> Vec<u8> {
+ assert!(
+ artifacts.len() <= STATE_MANIFEST_ENTRIES_MAX as usize,
+ "state manifest entry count {} exceeds {STATE_MANIFEST_ENTRIES_MAX}",
+ artifacts.len()
+ );
+ let mut out = Vec::with_capacity(9 + artifacts.len() * STATE_MANIFEST_ENTRY_LEN as usize + 8);
+ out.extend_from_slice(&STATE_MANIFEST_MAGIC);
+ out.extend_from_slice(&(artifacts.len() as u32).to_le_bytes());
+ out.push(STATE_MANIFEST_ENTRY_LEN);
+ for artifact in artifacts {
+ out.push(artifact.kind);
+ out.extend_from_slice(&artifact.frontier.to_le_bytes());
+ out.extend_from_slice(&artifact.len.to_le_bytes());
+ out.extend_from_slice(&artifact.checksum.to_le_bytes());
+ }
+ let mut hasher = XxHash3_64::new();
+ hasher.write(&out);
+ out.extend_from_slice(&hasher.finish().to_le_bytes());
+ out
+}
+
+/// Decode a manifest encoded by [`encode_state_manifest`].
+///
+/// Entries longer than [`STATE_MANIFEST_ENTRY_LEN`] decode fine: the known
+/// prefix is read and the remainder skipped, so a future encoder can append
+/// per-entry fields without breaking this decoder.
+///
+/// # Errors
+/// [`StateManifestError`] on truncation, magic/checksum mismatch, an entry
+/// count past the ceiling, or a shrunken entry length.
+///
+/// # Panics
+/// Unreachable: slice-to-array conversions are length-checked first.
+pub fn decode_state_manifest(bytes: &[u8]) -> Result<Vec<StateArtifact>, StateManifestError> {
+ const TRAILER: usize = size_of::<u64>();
+ let content_len = bytes
+ .len()
+ .checked_sub(TRAILER)
+ .ok_or(StateManifestError::Truncated)?;
+ let (content, trailer) = bytes.split_at(content_len);
+ let expected = u64::from_le_bytes(trailer.try_into().expect("trailer is 8 bytes"));
+ let mut hasher = XxHash3_64::new();
+ hasher.write(content);
+ let actual = hasher.finish();
+ if expected != actual {
+ return Err(StateManifestError::ChecksumMismatch { expected, actual });
+ }
+
+ let mut reader = ManifestReader { bytes: content };
+ if reader.take(STATE_MANIFEST_MAGIC.len())? != STATE_MANIFEST_MAGIC {
+ return Err(StateManifestError::BadMagic);
+ }
+ let count = reader.u32()?;
+ if count > STATE_MANIFEST_ENTRIES_MAX {
+ return Err(StateManifestError::TooManyEntries { count });
+ }
+ let entry_len = reader.u8()?;
+ if entry_len < STATE_MANIFEST_ENTRY_LEN {
+ return Err(StateManifestError::EntryTooShort { entry_len });
+ }
+
+ let mut artifacts = Vec::with_capacity(count as usize);
+ for _ in 0..count {
+ let entry = reader.take(entry_len as usize)?;
+ let mut entry = ManifestReader { bytes: entry };
+ artifacts.push(StateArtifact {
+ kind: entry.u8()?,
+ frontier: entry.u64()?,
+ len: entry.u64()?,
+ checksum: entry.u64()?,
+ });
+ }
+ if !reader.bytes.is_empty() {
+ return Err(StateManifestError::Truncated);
+ }
+ Ok(artifacts)
+}
+
+/// Little-endian cursor over the encoded manifest content.
+struct ManifestReader<'a> {
Review Comment:
`ManifestReader` here and `CodecReader` in client_table.rs are the same LE cursor (take/u8/u32/u64) duplicated within one crate, and the two codecs open-code the `XxHash3_64` new/write/finish triple four times when `state_artifact_checksum` already exists. one shared cursor (common `Truncated` error + `From` into each codec's enum) plus routing the hashing through the existing fn drops ~55 lines.
##########
core/binary_protocol/src/consensus/header.rs:
##########
@@ -1309,6 +1309,277 @@ impl ConsensusHeader for RepairRangeReplyHeader {
}
}
+// State transfer: descriptor + chunk pull frames.
+//
+// Plane-agnostic: the descriptor's BODY carries a state manifest (see the
+// consensus crate's `state_manifest`) listing N artifacts, and the chunk
+// frames address bytes by `(manifest index, offset)`. The metadata plane
+// ships two artifacts (snapshot + client table); the partition plane ships
+// its own set (segment logs, offsets) through the same frames.
+
+// RequestStateTransferHeader - restarted replica -> current primary
+
+/// Ask the current primary for a state-transfer target descriptor.
+///
+/// Answered with `StateTransferTarget`. Sent by a restarted replica after it
+/// adopts a view from a live primary; `nonce` correlates the whole transfer
+/// session.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct RequestStateTransferHeader {
+ pub checksum: u128,
+ pub checksum_body: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub view: u32,
+ pub release: u32,
+ pub command: Command2,
+ pub replica: u8,
+ pub reserved_frame: [u8; 66],
+
+ pub nonce: u128,
+ pub namespace: u64,
+ pub reserved: [u8; 104],
+}
+const _: () = {
+ assert!(size_of::<RequestStateTransferHeader>() == HEADER_SIZE);
+ assert!(
+ offset_of!(RequestStateTransferHeader, nonce)
+ == offset_of!(RequestStateTransferHeader, reserved_frame) + size_of::<[u8; 66]>()
+ );
+ assert!(
+ offset_of!(RequestStateTransferHeader, reserved) + size_of::<[u8; 104]>() == HEADER_SIZE
+ );
+};
+
+impl ConsensusHeader for RequestStateTransferHeader {
+ const COMMAND: Command2 = Command2::RequestStateTransfer;
+ fn operation(&self) -> Operation {
+ Operation::Reserved
+ }
+ fn command(&self) -> Command2 {
+ self.command
+ }
+ fn size(&self) -> u32 {
+ self.size
+ }
+
+ fn validate(&self) -> Result<(), ConsensusError> {
+ if self.command != Command2::RequestStateTransfer {
+ return Err(ConsensusError::InvalidCommand {
+ expected: Command2::RequestStateTransfer,
Review Comment:
the new header-only frames (`RequestStateTransferHeader`, `RequestStateChunkHeader`) don't size-check in `validate` while `EvictionHeader` does, and `StateTransferTargetHeader` constrains size only on the `available == 0` branch. the generic `Message::try_from` bound makes this safe either way, but adding `== HEADER_SIZE` to the two header-only frames (and a comment that the `available == 1` body is bounded downstream by the manifest caps) keeps the validate surface consistent.
##########
core/server-ng/src/responses.rs:
##########
@@ -1340,6 +1340,14 @@ pub(crate) fn build_raw_pat_reply(
return Ok(committed);
}
let header_len = std::mem::size_of::<ReplyHeader>();
+ // A `Reply` whose result section is nonzero is not a successful commit but a
+ // committed business rejection or a `TransientNotCommitted` retry frame, both
+ // with no payload and no token to ship. Pass it through so the client decodes
+ // the typed result (and, for a transient, replays) instead of having a raw
+ // token grafted onto a rejection body.
+ if result_code(&committed.as_slice()[header_len..]) != Some(0) {
Review Comment:
this early return is dominated by the existing `result_code` gate a few lines down - same leading bytes, and the pre-existing one is better bounded (`get(header_len..size)` vs this unbounded slice that can read past the declared size into buffer padding). every frame this catches, that one catches. if this was added for an observed bug, the old gate already covered that case, which would mean the actual bug is elsewhere - worth a note on what motivated it. otherwise just drop this one.
##########
core/integration/src/harness/handle/server.rs:
##########
@@ -978,6 +999,35 @@ impl Restartable for ServerHandle {
}
}
+impl ServerHandle {
+ /// Stop this node, delete its data directory, then start it again -- a
+ /// node that rejoins the cluster with no on-disk history, as a fresh
+ /// operator-provisioned replacement or a wiped disk does.
+ ///
+ /// Distinct from [`Restartable::restart`], which preserves the WAL and so
+ /// exercises the recovery-with-history path. The empty data directory is
+ /// what forces the state-transfer join: local recovery has nothing, and
+ /// the committed prefix the node missed no longer exists as WAL entries on
+ /// the peers that checkpointed it.
+ pub fn restart_from_clean_slate(&mut self) -> Result<(), TestBinaryError> {
+ let cleanup = self.config.cleanup;
Review Comment:
the cleanup save/restore dance is dead - `ServerConfig::cleanup` is never read for behavior anywhere in the harness (the real gate is `TestContext::cleanup`), and this fn wipes the data dir itself two lines later. copied from the same dead pattern in `restart`.
##########
core/shard/src/lib.rs:
##########
@@ -2035,6 +2179,40 @@ where
if planes.0.persist_superblock_if_needed(consensus).await {
dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
}
+ // State transfer (rejoin behind the peers' retained floor): the
+ // adopted view names a live primary to fetch snapshot-shaped state
+ // from. The commit walk and journal repair are deferred until the
+ // install lands -- walking the pre-transfer STM would apply ops the
+ // snapshot already contains, and the transfer replaces the table
+ // anyway.
+ //
+ // Outside the superblock gate above: that gate fail-closes the VSR
+ // actions this replica would VOUCH with (notably `PrepareOk`) until
+ // the adopted view is durable. Requesting a transfer vouches for
+ // nothing -- it only pulls state -- and a transferring replica
+ // withholds `PrepareOk` on its own (`is_transferring`). Gating it
+ // would also wedge the one path that repairs a replica whose gap
+ // sits below every peer's floor.
+ if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
Review Comment:
this arms on stage alone and ignores whether `handle_start_view` accepted anything - it returns an empty action set for wrong-primary / old-view / below-floor, but the arm still mints a fresh nonce, zeroes `attempts`, and fires a request at a possibly-non-primary. it self-heals via `available = 0`, but it resets the retry budget, and a stream of stale StartViews re-mints the nonce each time so the in-flight descriptor gets dropped by the nonce filter. gate on the StartView actually being adopted.
##########
core/integration/tests/data_integrity/mod.rs:
##########
@@ -15,13 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-// Requires state transfer: the 5 MB bench fill is thousands of ops while the
-// partition journal's evicted ring retains only the last 4096, so a restarted
-// replica's rejoin window exceeds what journal repair can serve. The commit
-// floor lets recovered segments stand in for the evicted prefix, but the
-// sub-floor stats/offset seeding this test asserts (exact messages_count /
-// size_bytes across the restart) is state transfer's job.
-#[cfg(not(feature = "vsr"))]
+// Partially vsr-gated inside the module: the bench-fill test requires
+// PARTITION-plane state transfer (sub-floor stats/offset seeding), which is
+// not implemented yet; the metadata-only deletion/restart test runs under
Review Comment:
dropping the module-level gate puts `should_handle_resource_deletion_and_restart` into the vsr lane for the first time, and CI doesn't run that lane - green here doesn't prove it passes. worth pasting a local `--features vsr` run of it in the PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org