This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch partition_superblock
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/partition_superblock by this
push:
new 78f6d9da9 addres review comments
78f6d9da9 is described below
commit 78f6d9da9b3634fb2622970ef9ec637f8a0e2fd3
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Thu Aug 6 07:26:23 2026 +0200
addres review comments
---
core/binary_protocol/src/consensus/header.rs | 19 +-
core/configs/src/server_ng_config/defaults.rs | 5 +
core/configs/src/server_ng_config/partition.rs | 29 +
core/consensus/src/impls.rs | 3 +
core/consensus/src/state_transfer.rs | 53 +-
core/consensus/src/vsr_state.rs | 24 +-
.../server/scenarios/purge_delete_scenario.rs | 18 +-
core/journal/src/local_gate.rs | 97 +++
core/metadata/src/impls/recovery.rs | 1 +
core/metadata/src/lib.rs | 4 +
core/partitions/src/iggy_partition.rs | 87 ++-
core/partitions/src/journal.rs | 36 +-
core/partitions/src/state_transfer.rs | 344 +++++++++--
core/server-ng/config.toml | 14 +
core/server-ng/src/bootstrap.rs | 57 +-
core/server-ng/src/partition_helpers.rs | 52 +-
core/server-ng/src/segment_recovery.rs | 71 ++-
core/server-ng/src/server_error.rs | 2 +-
core/shard/src/lib.rs | 662 ++++++++++++++-------
core/shard/src/metrics.rs | 14 +
core/shard/src/router.rs | 23 +-
21 files changed, 1276 insertions(+), 339 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs
b/core/binary_protocol/src/consensus/header.rs
index 12213676b..e71f1004c 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1416,11 +1416,14 @@ pub struct StateTransferTargetHeader {
pub namespace: u64,
pub available: u8,
/// Set on an `available == 0` refusal that means "not right now" rather
than
- /// "this node is broken": the requester then re-arms on a flat interval
- /// instead of charging its consecutive-failure count, whose exponential
- /// backoff climbs to 1024x the retry interval and is reset only by a
- /// completed install. A serving primary momentarily behind its own
frontier
- /// is the common case under produce load.
+ /// "this node is broken".
+ ///
+ /// PARTITION arm only: it is the only side with a consecutive-failure
count
+ /// to charge. The requester then re-arms on a flat interval instead of
+ /// charging that count, whose exponential backoff climbs to 1024x the
retry
+ /// interval and is reset only by a completed install. A serving primary
+ /// momentarily behind its own frontier is the common case under produce
+ /// load.
///
/// This and `commit_max` below claim the HEAD of what used to be the
/// reserved tail, so every pre-existing field keeps its published offset:
@@ -1434,7 +1437,11 @@ pub struct StateTransferTargetHeader {
pub reserved_alignment: [u8; 6],
/// Serving replica's `commit_max` when the descriptor was built.
///
- /// A receiver refuses an offer from a replica that knows LESS than it
does:
+ /// Read by the PARTITION receiver only; the metadata arm branches on
+ /// `available` and falls back to journal repair without a refusal.
+ ///
+ /// A partition receiver refuses an offer from a replica that knows LESS
+ /// than it does:
/// without this the descriptor carried no proof of the sender's own
/// progress, and a phantom view-0 primary (a group whose directory
vanished
/// boots `init()` rather than `init_as_backup()`, comes up Normal at view
0,
diff --git a/core/configs/src/server_ng_config/defaults.rs
b/core/configs/src/server_ng_config/defaults.rs
index e68f0889c..40695a63e 100644
--- a/core/configs/src/server_ng_config/defaults.rs
+++ b/core/configs/src/server_ng_config/defaults.rs
@@ -176,6 +176,11 @@ impl Default for PartitionConfig {
prepare_queue_depth: partition.prepare_queue_depth as usize,
evicted_ring_capacity: partition.evicted_ring_capacity as usize,
evicted_ring_bytes_max:
partition.evicted_ring_bytes_max.parse().unwrap(),
+ transfer_served_cache_bytes_max: partition
+ .transfer_served_cache_bytes_max
+ .parse()
+ .unwrap(),
+ transfer_artifact_bytes_max:
partition.transfer_artifact_bytes_max.parse().unwrap(),
}
}
}
diff --git a/core/configs/src/server_ng_config/partition.rs
b/core/configs/src/server_ng_config/partition.rs
index 580668e4b..66d428fd7 100644
--- a/core/configs/src/server_ng_config/partition.rs
+++ b/core/configs/src/server_ng_config/partition.rs
@@ -95,6 +95,27 @@ pub struct PartitionConfig {
/// [`MAX_EVICTED_RING_BYTES`].
#[config_env(leaf)]
pub evicted_ring_bytes_max: IggyByteSize,
+
+ /// Byte budget for segment payloads a SERVING shard keeps resident to
+ /// answer state-transfer chunk requests, per shard (so the process-wide
+ /// bound is this times the shard count).
+ ///
+ /// Sized for concurrent pulls, not one: at exactly one maximum segment a
+ /// single receiver arming several transfers thrashes the cache by itself,
+ /// and every miss re-reads and re-hashes a whole segment to serve one
+ /// chunk. Must be > 0.
+ #[config_env(leaf)]
+ pub transfer_served_cache_bytes_max: IggyByteSize,
+
+ /// Alloc ceiling for ONE received state-transfer artifact, per shard.
+ ///
+ /// A receiver holds the whole artifact resident through verify, walk and
+ /// staging write, so the in-flight cap multiplies this. It must stay above
+ /// the largest legal segment (`segment.size` plus the one batch a segment
+ /// may overshoot it by) or legal segments are rejected deterministically.
+ /// Must be > 0.
+ #[config_env(leaf)]
+ pub transfer_artifact_bytes_max: IggyByteSize,
}
impl Validatable<ConfigurationError> for PartitionConfig {
@@ -121,6 +142,14 @@ impl Validatable<ConfigurationError> for PartitionConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if self.transfer_served_cache_bytes_max.as_bytes_u64() == 0 {
+ eprintln!("{COMPONENT_NG}
partition.transfer_served_cache_bytes_max must be > 0");
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if self.transfer_artifact_bytes_max.as_bytes_u64() == 0 {
+ eprintln!("{COMPONENT_NG} partition.transfer_artifact_bytes_max
must be > 0");
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
let ring_bytes = self.evicted_ring_bytes_max.as_bytes_u64();
if ring_bytes == 0 {
eprintln!("{COMPONENT_NG} partition.evicted_ring_bytes_max must be
> 0");
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index dda3fa9a8..d23378f15 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -1527,6 +1527,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
commit_max: self.commit_max.get(),
checkpoint_op,
checkpoint_checksum,
+ // Consensus mints no message offsets: the PARTITION plane stamps
+ // this in before it writes (`IggyPartition::write_superblock`).
+ offset_frontier: 0,
}
}
diff --git a/core/consensus/src/state_transfer.rs
b/core/consensus/src/state_transfer.rs
index 0a1ff3768..f5cf14e8e 100644
--- a/core/consensus/src/state_transfer.rs
+++ b/core/consensus/src/state_transfer.rs
@@ -39,7 +39,10 @@ pub const STATE_TRANSFER_MAX_STALL_RETRIES: u32 = 5;
/// Decode-failure rounds a receiver spends on ONE offered generation
/// before refusing to pull it again.
///
-/// METADATA plane only, keyed on `snapshot_seq`: a peer whose snapshot
+/// Read by the METADATA arm only -- it lives here because the chunk cursor it
+/// pairs with is plane-agnostic, not because both planes use it.
+///
+/// Keyed on `snapshot_seq`: a peer whose snapshot
/// generation advances resets the budget (new bytes are worth full
/// retries), while a generation this build cannot decode costs one refused
/// descriptor per repair round instead of a full pull. The partition plane
@@ -60,13 +63,6 @@ pub struct ArtifactProgress {
pub buf: Vec<u8>,
}
-impl ArtifactProgress {
- #[must_use]
- pub const fn complete(&self) -> bool {
- self.buf.len() as u64 == self.entry.len
- }
-}
-
/// What [`next_pending_chunk`] and [`append_chunk`] need from one slot.
///
/// Exists so a plane can track completion in richer shapes -- the
@@ -79,6 +75,11 @@ pub trait ChunkProgress {
/// Only called after the cursor checks `received + payload <= declared`,
/// so an impl whose slot cannot grow (already complete) never sees it.
fn extend_from_chunk(&mut self, payload: &[u8]);
+ /// Reserve room for the whole declared length, called once per artifact on
+ /// its FIRST chunk (see [`append_chunk`]). Default: nothing, for slots
that
+ /// do not accumulate in memory. Reserving at accept time instead would
+ /// commit address space for every manifest entry at once.
+ fn reserve_declared(&mut self) {}
fn complete(&self) -> bool {
self.received_len() == self.declared_len()
}
@@ -94,18 +95,16 @@ impl ChunkProgress for ArtifactProgress {
}
fn extend_from_chunk(&mut self, payload: &[u8]) {
- // Reserved on the FIRST chunk of this artifact, not when the manifest
is
- // accepted: reserving every artifact up front is an eager
address-space
- // commit of the whole manifest, and a receiver only ever pulls one
- // artifact at a time. Exact rather than geometric -- `entry.len`
already
- // passed the caller's per-kind caps, and doubling to gigabyte sizes
would
- // copy roughly twice the bytes at a ~1.5x transient peak.
- if self.buf.is_empty() {
- #[allow(clippy::cast_possible_truncation)]
- self.buf.reserve_exact(self.entry.len as usize);
- }
self.buf.extend_from_slice(payload);
}
+
+ fn reserve_declared(&mut self) {
+ // Exact rather than geometric: `entry.len` already passed the caller's
+ // per-kind caps, and doubling to gigabyte sizes copies roughly twice
the
+ // bytes at a ~1.5x transient peak.
+ #[allow(clippy::cast_possible_truncation)]
+ self.buf.reserve_exact(self.entry.len as usize);
+ }
}
/// Next `(artifact index, offset, len)` to request.
@@ -138,13 +137,12 @@ pub fn next_pending_chunk<T: ChunkProgress>(
/// lockstep, so anything else is a duplicate or reorder -- the stall retry
/// re-requests from the current frontier), an overrun past the declared
/// length, and a zero-byte payload. The first-incomplete restriction mirrors
-/// what [`next_pending_chunk`] would have requested: without it a peer that
-/// pushes one byte into EVERY manifest entry makes each artifact's
-/// first-chunk `reserve_exact` fire, committing address space for the sum of
-/// all declared lengths at once -- the whole-manifest reservation the
-/// receiver deliberately refuses to make up front, and a failed `Vec`
-/// reservation aborts the process rather than erroring. A zero-byte payload
-/// is not progress: it extends nothing and the same offset is re-requested
+/// what [`next_pending_chunk`] would have requested, and bounds the
+/// reservation below to ONE artifact at a time: without it a peer that pushes
+/// one byte into every manifest entry would make each slot reserve its whole
+/// declared length, committing address space for the sum of the manifest, and
+/// a failed `Vec` reservation aborts the process rather than erroring. A
+/// zero-byte payload is not progress: it extends nothing and the same offset
is re-requested
/// immediately, and resetting liveness counters on one is what turned a
/// short rebuilt offer into an unbounded empty-frame ping-pong on the
/// metadata plane. The serving side refuses to produce these now; the guard
@@ -177,6 +175,11 @@ pub fn append_chunk<T: ChunkProgress>(
if payload.is_empty() {
return false;
}
+ // First chunk of this artifact: give the slot its full declared length in
+ // one allocation, so a segment-sized artifact is not grown by doubling.
+ if artifact.received_len() == 0 {
+ artifact.reserve_declared();
+ }
artifact.extend_from_chunk(payload);
true
}
diff --git a/core/consensus/src/vsr_state.rs b/core/consensus/src/vsr_state.rs
index 6523f38dc..bdef87c32 100644
--- a/core/consensus/src/vsr_state.rs
+++ b/core/consensus/src/vsr_state.rs
@@ -31,8 +31,8 @@ use std::fmt;
/// Number of bytes [`VsrState::to_bytes`] produces and [`VsrState::try_from`]
/// expects: `cluster`(16) + `replica_id`(1) + `replica_count`(1) + `view`(4)
/// + `log_view`(4) + `commit_max`(8) + `checkpoint_op`(8)
-/// + `checkpoint_checksum`(16).
-pub const ENCODED_LEN: usize = 58;
+/// + `checkpoint_checksum`(16) + `offset_frontier`(8).
+pub const ENCODED_LEN: usize = 66;
/// The durable consensus state of one replica for one consensus group.
///
@@ -69,6 +69,22 @@ pub struct VsrState {
/// Integrity tag of the paired checkpoint, detecting a torn
/// snapshot/superblock pairing across a crash.
pub checkpoint_checksum: u128,
+ /// PARTITION plane: the next message offset this replica will mint, or `0`
+ /// for a group whose offset space is still empty.
+ ///
+ /// A durable LOWER BOUND, not a completeness claim: boot takes the max of
+ /// this and whatever the recovered segments prove. It exists because
+ /// nothing else durably names the frontier once the segments that carried
+ /// it are gone -- a state-transfer install of an all-GC'd origin, a crash
+ /// inside the install's swap window, and the fence-and-rebuild path all
+ /// leave a replica whose counter would otherwise restart at 0 while the
+ /// group is at N. That is not a lag: replicas re-stamp `base_offset` from
+ /// this counter and recompute `batch_checksum` over it, so the next
+ /// replicated prepare would persist different bytes here than on every
+ /// peer, silently.
+ ///
+ /// Always `0` on the metadata plane, which mints no message offsets.
+ pub offset_frontier: u64,
}
impl VsrState {
@@ -84,6 +100,7 @@ impl VsrState {
out[26..34].copy_from_slice(&self.commit_max.to_le_bytes());
out[34..42].copy_from_slice(&self.checkpoint_op.to_le_bytes());
out[42..58].copy_from_slice(&self.checkpoint_checksum.to_le_bytes());
+ out[58..66].copy_from_slice(&self.offset_frontier.to_le_bytes());
out
}
}
@@ -108,6 +125,7 @@ impl TryFrom<&[u8]> for VsrState {
commit_max: u64::from_le_bytes(field(bytes, 26)),
checkpoint_op: u64::from_le_bytes(field(bytes, 34)),
checkpoint_checksum: u128::from_le_bytes(field(bytes, 42)),
+ offset_frontier: u64::from_le_bytes(field(bytes, 58)),
};
// A record violating `log_view <= view` decodes into a replica that
looks
// healthy locally while `DoViewChangeHeader::validate` makes every
peer drop
@@ -178,6 +196,7 @@ mod tests {
commit_max: 6,
checkpoint_op: 7,
checkpoint_checksum: 8,
+ offset_frontier: 0,
};
let bytes = state.to_bytes();
assert_eq!(bytes.len(), ENCODED_LEN);
@@ -209,6 +228,7 @@ mod tests {
commit_max: 0,
checkpoint_op: 0,
checkpoint_checksum: 0,
+ offset_frontier: 0,
}
.to_bytes();
bytes[22] = 5; // log_view = 5, view stays 4
diff --git a/core/integration/tests/server/scenarios/purge_delete_scenario.rs
b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
index 97cf1c816..606002701 100644
--- a/core/integration/tests/server/scenarios/purge_delete_scenario.rs
+++ b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
@@ -994,12 +994,18 @@ pub async fn run_purge_topic(harness: &mut TestHarness,
restart_server: bool) {
// files in the SAME frame that plants the layout, so the instant assert is
// correct there and strictly stronger; a poll would hide a regression that
// clears them one frame late.
- let offsets_deadline = std::time::Instant::now()
- + if restart_server {
- std::time::Duration::from_secs(10)
- } else {
- std::time::Duration::ZERO
- };
+ // vsr-only, and only for the restart cells: the legacy flavor purges
+ // synchronously, and without a restart the pump clears offsets and files
in
+ // the SAME frame that plants the layout, so the instant assert is correct
+ // and strictly stronger there. Kept short -- a client-visible stale offset
+ // after purge-then-restart is a real (bounded) window, not something to
+ // paper over with a long tolerance.
+ let poll_window = if cfg!(feature = "vsr") && restart_server {
+ std::time::Duration::from_secs(2)
+ } else {
+ std::time::Duration::ZERO
+ };
+ let offsets_deadline = std::time::Instant::now() + poll_window;
loop {
let consumer_offset = client
.get_consumer_offset(&consumer, &stream_ident, &topic_ident,
Some(PARTITION_ID))
diff --git a/core/journal/src/local_gate.rs b/core/journal/src/local_gate.rs
index 983ac51bf..6d2b7b86e 100644
--- a/core/journal/src/local_gate.rs
+++ b/core/journal/src/local_gate.rs
@@ -102,3 +102,100 @@ impl Drop for LocalGateGuard<'_> {
}
}
}
+
+// NOT behind `cfg(debug_assertions)`: release is exactly the build where this
+// gate is the only enforcement of the superblock single-writer contract (the
+// `WritingGuard` tripwire is debug-only, `write` takes `&self`, and two
+// overlapping writers collide on one fixed `.tmp` path and tear a slot while
+// both return `Ok`).
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use futures::FutureExt;
+ use futures::future::join;
+ use std::rc::Rc;
+
+ /// Two writers queued on the same gate run one after the other, never
+ /// interleaved -- the property a torn superblock slot depends on.
+ #[compio::test]
+ async fn given_two_waiters_when_acquiring_should_serialize() {
+ let gate = LocalGate::new();
+ let gate = &gate;
+ let log: Rc<RefCell<Vec<&'static str>>> =
Rc::new(RefCell::new(Vec::new()));
+
+ let first = {
+ let log = Rc::clone(&log);
+ async move {
+ let guard = gate.acquire().await;
+ log.borrow_mut().push("first:enter");
+ // A yield inside the critical section: without exclusion the
+ // second writer would slot in right here.
+ compio::time::sleep(std::time::Duration::from_millis(1)).await;
+ log.borrow_mut().push("first:exit");
+ drop(guard);
+ }
+ };
+ let second = {
+ let log = Rc::clone(&log);
+ async move {
+ let guard = gate.acquire().await;
+ log.borrow_mut().push("second:enter");
+ compio::time::sleep(std::time::Duration::from_millis(1)).await;
+ log.borrow_mut().push("second:exit");
+ drop(guard);
+ }
+ };
+ join(first, second).await;
+
+ let log = log.borrow();
+ assert_eq!(log.len(), 4, "both sections ran: {log:?}");
+ let first_exit = log.iter().position(|entry| *entry ==
"first:exit").unwrap();
+ let second_enter = log
+ .iter()
+ .position(|entry| *entry == "second:enter")
+ .unwrap();
+ assert!(
+ first_exit < second_enter,
+ "sections must not interleave: {log:?}"
+ );
+ }
+
+ /// Dropping the guard wakes a queued waiter, so the gate does not wedge
+ /// once contended.
+ #[compio::test]
+ async fn given_queued_waiter_when_guard_drops_should_wake_it() {
+ let gate = LocalGate::new();
+ let held = gate.acquire().await;
+
+ let mut waiter = Box::pin(gate.acquire());
+ assert!(
+ waiter.as_mut().now_or_never().is_none(),
+ "the gate is held, so the waiter must park"
+ );
+
+ drop(held);
+ assert!(
+ waiter.now_or_never().is_some(),
+ "dropping the guard must wake the queued waiter"
+ );
+ }
+
+ /// A waiter that goes away leaves only a stale waker behind: the next
+ /// acquirer still gets the gate. Cancel safety is load-bearing here, since
+ /// every acquire site sits in a future a shard can drop.
+ #[compio::test]
+ async fn given_dropped_waiter_when_guard_releases_should_not_wedge() {
+ let gate = LocalGate::new();
+ let held = gate.acquire().await;
+
+ let mut abandoned = Box::pin(gate.acquire());
+ assert!(abandoned.as_mut().now_or_never().is_none());
+ drop(abandoned);
+
+ drop(held);
+ assert!(
+ gate.acquire().now_or_never().is_some(),
+ "a dropped waiter must not keep the gate busy"
+ );
+ }
+}
diff --git a/core/metadata/src/impls/recovery.rs
b/core/metadata/src/impls/recovery.rs
index 0ed4cfc19..286d286e2 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -781,6 +781,7 @@ mod tests {
commit_max: 100,
checkpoint_op,
checkpoint_checksum,
+ offset_frontier: 0,
}
}
diff --git a/core/metadata/src/lib.rs b/core/metadata/src/lib.rs
index 5c4ba390e..4d4d24003 100644
--- a/core/metadata/src/lib.rs
+++ b/core/metadata/src/lib.rs
@@ -27,5 +27,9 @@ pub use impls::metadata::{
apply_committed_prepare,
};
+// Recovery vocabulary other crates name in their own signatures and error
+// enums, so they do not have to spell the `impls::` path.
+pub use impls::recovery::{IdentityField, RecoveryError, ReplicaIdentity};
+
// Re-export MuxStateMachine for use in other modules
pub use stm::mux::MuxStateMachine;
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 16899e94f..85ceae6de 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -200,6 +200,10 @@ where
/// at network round-trip rate. Reset only by
/// [`Self::note_transfer_installed`]; drives the re-arm backoff.
transfer_failures: u32,
+ /// CONSECUTIVE transient refusals (a peer that cannot serve right now).
+ /// Drives log escalation only -- never the backoff. See
+ /// [`Self::record_transfer_refusal`].
+ transfer_refusals: u32,
/// A scheduled transfer re-arm: try `peer` again once `after_ticks`
/// consensus ticks elapse. Owned by the shard tick sweep; while one is
/// pending, the repair-refusal trigger must not arm concurrently.
@@ -364,6 +368,7 @@ where
transfer: None,
transfer_attempts: 0,
transfer_failures: 0,
+ transfer_refusals: 0,
transfer_rearm: None,
segment_checksum_cache:
RefCell::new(std::collections::HashMap::new()),
reuse_scan_memo: RefCell::new(None),
@@ -476,7 +481,8 @@ where
if !self.consensus.needs_superblock_persist() {
return true;
}
- self.write_superblock(superblock.as_ref()).await
+ self.write_superblock(superblock.as_ref(), self.offset_frontier())
+ .await
}
/// Write the current VSR state under [`Self::superblock_lock`].
@@ -494,14 +500,23 @@ where
/// send for this group, goes quiet, and its peers elect around it. Only
/// THIS partition's group is fenced; the rest of the node keeps serving.
#[allow(clippy::future_not_send)]
- async fn write_superblock(&self, superblock: &SB) -> bool {
+ async fn write_superblock(&self, superblock: &SB, offset_frontier: u64) ->
bool {
// The pairing fields stay `(0, 0)` and `commit_max` is a dead write
// on this plane: nothing reads either back (`restore_partition_view`
// restores view/log_view only), because recovery re-derives the
// install floor from the installed segments at boot -- a crash after
// an install does not re-run the transfer. Written anyway so the
// record shape matches the metadata plane's.
- let state = self.consensus.vsr_state(0, 0);
+ //
+ // `offset_frontier` is NOT dead: it is the only durable carrier of the
+ // group's offset space once the segments that named it are gone. Every
+ // write stamps the current counter, so whichever write lands last (a
+ // view change, or the explicit persist an install issues) leaves a
+ // lower bound boot can re-seed from.
+ let mut state = self.consensus.vsr_state(0, 0);
+ // Never regresses: a caller recording an incoming frontier passes a
+ // larger value, and the ordinary gate passes the live counter.
+ state.offset_frontier = offset_frontier.max(self.offset_frontier());
match superblock.write(&state.to_bytes()).await {
Ok(()) => {
self.consensus
@@ -541,6 +556,53 @@ where
}
}
+ /// The next message offset this replica will mint, `0` while the offset
+ /// space is still empty. The value stamped into the durable record.
+ #[must_use]
+ pub fn offset_frontier(&self) -> u64 {
+ if self.should_increment_offset {
+ self.offset.load(Ordering::Acquire).saturating_add(1)
+ } else {
+ 0
+ }
+ }
+
+ /// Force the durable record to catch up with the current offset frontier,
+ /// outside the view-change gate.
+ ///
+ /// [`Self::persist_superblock_if_needed`] fires on `(view, log_view)`
+ /// changes only, which is the right trigger for the split-brain fence and
+ /// the wrong one for the frontier: an install can move the counter by
+ /// millions without touching the view. Called where the frontier changes
+ /// with nothing else durable naming it -- after a state-transfer install
+ /// and after the convergence that follows a failed one. Returns whether
the
+ /// record now holds it; a failure is logged by the writer and left to the
+ /// ordinary retry, since the install itself already succeeded.
+ #[allow(clippy::future_not_send)]
+ pub async fn persist_offset_frontier(&self) -> bool {
+ self.persist_offset_frontier_at(self.offset_frontier())
+ .await
+ }
+
+ /// [`Self::persist_offset_frontier`] for a frontier this replica has not
+ /// reached yet.
+ ///
+ /// Used to record an INCOMING frontier before a destructive swap: the
+ /// install unlinks the old chain and fsyncs that before the first staged
+ /// rename lands, and boot sweeps `.log.staging` unconditionally, so a
crash
+ /// in that window otherwise leaves no copy of the frontier anywhere.
Writing
+ /// the claim first makes it a durable lower bound the whole way through,
and
+ /// over-claiming is harmless: the convergence that follows a failed
install
+ /// seeds the counter from the same artifact frontier.
+ #[allow(clippy::future_not_send)]
+ pub async fn persist_offset_frontier_at(&self, frontier: u64) -> bool {
+ let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
+ return true;
+ };
+ let _superblock_guard = self.superblock_lock.acquire().await;
+ self.write_superblock(superblock.as_ref(), frontier).await
+ }
+
/// Burn one transfer stall round; `true` once the budget is exhausted.
/// Lives on the partition, not the session, so a re-minted session
/// cannot reset it (see [`Self::transfer_attempts`]).
@@ -572,6 +634,19 @@ where
/// consecutive-failure count.
pub const fn note_transfer_installed(&mut self) {
self.transfer_failures = 0;
+ self.transfer_refusals = 0;
+ }
+
+ /// Charge one TRANSIENT refusal and return the consecutive count.
+ ///
+ /// Separate from [`Self::record_transfer_failure`] on purpose: a transient
+ /// refusal must not touch the exponential backoff (the flat retry interval
+ /// is the point), but a partition refused for hours still has to be
+ /// visible, so the count exists only to escalate logging and feed a
metric.
+ /// Reset by [`Self::note_transfer_installed`] alongside the failure count.
+ pub const fn record_transfer_refusal(&mut self) -> u32 {
+ self.transfer_refusals = self.transfer_refusals.saturating_add(1);
+ self.transfer_refusals
}
/// A fresh re-arm is scheduled: the stall budget starts over for it.
@@ -3432,6 +3507,12 @@ where
// Recreate a fresh empty segment at offset 0 with real writers.
let start_offset = 0u64;
self.install_empty_segment(config, start_offset).await?;
+ // Make the unlinks AND the replanted dirent durable together: without
+ // this a crash can resurrect pre-purge segments until the boot
re-purge
+ // fires. Bounded and self-healing, but the fsync is one call.
+ if let Some(partition_dir) = self.partition_dir.clone() {
+ let _ = crate::state_transfer::fsync_dir(&partition_dir).await;
+ }
// Reset the offset counters so new messages start at offset 0.
self.offset.store(start_offset, Ordering::Release);
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index 01487f66b..acc4f926a 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -24,7 +24,7 @@ use server_common::{
use std::io;
use std::{
cell::{Cell, UnsafeCell},
- collections::{BTreeMap, HashMap, HashSet, VecDeque},
+ collections::{BTreeMap, HashMap, VecDeque},
};
use tracing::warn;
@@ -682,13 +682,26 @@ where
pub fn repaired_window_shape(&self, floor: u64, to_op: u64) ->
RepairedWindowShape {
let headers = unsafe { &*self.headers.get() };
let expected = to_op.saturating_sub(floor);
- // At most one entry per in-window op can land, so the window bounds
the
- // hint: sizing it for every resident header allocated (and memset)
about
- // a megabyte of table per repair round on the floor-refusal path,
where
- // the header vec grows with the live tail.
+ // More in-window ops than resident headers can never be covered, and
+ // `expected` is unbounded here (`to_op` rides the local `commit_max`),
+ // so this is both the early answer and what keeps the bitset below
from
+ // being sized off an arbitrary number.
+ if expected > headers.len() as u64 {
+ return RepairedWindowShape {
+ complete: false,
+ holds_messages: headers.iter().any(|header| {
+ header.op > floor
+ && header.op <= to_op
+ && header.operation == Operation::SendMessages
+ }),
+ };
+ }
+ // Dense window, so a bitset beats a `HashSet`: no hashing per op and
one
+ // allocation of `expected / 8` bytes.
#[allow(clippy::cast_possible_truncation)]
- let capacity = expected.min(headers.len() as u64) as usize;
- let mut present: HashSet<u64> = HashSet::with_capacity(capacity);
+ let expected_len = expected as usize;
+ let mut present = vec![false; expected_len];
+ let mut covered = 0usize;
let mut holds_messages = false;
for header in headers
.iter()
@@ -697,11 +710,16 @@ where
if header.operation == Operation::SendMessages {
holds_messages = true;
}
- present.insert(header.op);
+ #[allow(clippy::cast_possible_truncation)]
+ let slot = (header.op - floor - 1) as usize;
+ if !present[slot] {
+ present[slot] = true;
+ covered += 1;
+ }
}
RepairedWindowShape {
// In-window ops only, deduplicated, so a count match IS coverage.
- complete: present.len() as u64 == expected,
+ complete: covered == expected_len,
holds_messages,
}
}
diff --git a/core/partitions/src/state_transfer.rs
b/core/partitions/src/state_transfer.rs
index ec1cbd03f..37ef73409 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -84,9 +84,10 @@ pub struct PartitionTransferSession {
pub nonce: u128,
/// Serving primary; also the stall re-request target.
pub peer: u8,
- /// Serving peer's applied frontier from the accepted descriptor. Doubles
- /// as the decode-budget generation: segments are append-only, so a new
- /// commit frontier genuinely means new bytes.
+ /// Serving peer's applied frontier from the accepted descriptor.
+ ///
+ /// Not a decode-budget generation: this plane keeps no such budget (see
the
+ /// struct doc), it counts consecutive failures on the partition instead.
pub commit_op: u64,
/// One slot per offered artifact, in manifest order. A slot moves from
/// `Pending` to `Staged` when its segment payload is validated and
@@ -161,12 +162,22 @@ impl consensus::ChunkProgress for TransferArtifact {
fn extend_from_chunk(&mut self, payload: &[u8]) {
match self {
- Self::Pending(progress) => progress.buf.extend_from_slice(payload),
+ // Delegated, not re-implemented: the two must agree about how a
+ // buffer grows, and the reservation below only fires if this arm
+ // routes through the same impl.
+ Self::Pending(progress) => progress.extend_from_chunk(payload),
// Unreachable through `append_chunk`: a staged slot reports
// itself complete, so no in-window offset can address it.
Self::Staged(_) => debug_assert!(false, "chunk appended to a
staged artifact"),
}
}
+
+ fn reserve_declared(&mut self) {
+ match self {
+ Self::Pending(progress) => progress.reserve_declared(),
+ Self::Staged(_) => {}
+ }
+ }
}
/// What the receiver learned walking one validated, staged segment artifact:
@@ -206,7 +217,9 @@ pub struct StagedSegmentMeta {
/// bytes" hold.
pub(crate) struct SegmentChecksumMemo {
hashed_len: u64,
- checksum: u64,
+ /// The stamp is NOT cached alongside: `StateArtifactHasher::finish` takes
+ /// `&self`, so it is a read of this hasher, and a second copy is just a
+ /// field that can drift.
hasher: consensus::state_manifest::StateArtifactHasher,
}
@@ -214,7 +227,6 @@ impl SegmentChecksumMemo {
fn new() -> Self {
Self {
hashed_len: 0,
- checksum: 0,
hasher: consensus::state_manifest::StateArtifactHasher::new(),
}
}
@@ -234,6 +246,31 @@ pub(crate) struct ReuseScanMemo {
adopted: Vec<(u32, StagedSegmentMeta)>,
}
+impl StagedSegmentMeta {
+ /// Assemble the metadata a completed walk produced. Shared by the spill
and
+ /// the reuse-adopt path, which differ only in whether they also wrote the
+ /// payload.
+ const fn from_walk(
+ entry: &consensus::StateArtifact,
+ stats: SegmentWalkStats,
+ index_size: u64,
+ log_staging: PathBuf,
+ index_staging: PathBuf,
+ ) -> Self {
+ Self {
+ start_offset: entry.frontier,
+ end_offset: stats.end_offset,
+ size: entry.len,
+ index_size,
+ start_timestamp: stats.start_timestamp,
+ end_timestamp: stats.end_timestamp,
+ max_timestamp: stats.max_timestamp,
+ log_staging,
+ index_staging,
+ }
+ }
+}
+
/// The consumer-offset artifact: both offset maps plus the applied purge
/// generation, at the offer's `commit_op`.
///
@@ -354,6 +391,15 @@ impl ConsumerOffsetsWire {
max: CONSUMER_OFFSETS_ENTRIES_MAX,
});
}
+ // The count is peer input and the reservation is 12 bytes per element
+ // after alignment, so it is checked against the bytes actually present
+ // before allocating: a ~30 byte artifact could otherwise ask for tens
of
+ // megabytes across the two sections. 12 is the wire stride below --
the
+ // groups call sees exactly `12 * count` bytes remaining, so a wider
+ // guard would reject every non-empty artifact.
+ if count as usize * (size_of::<u32>() + size_of::<u64>()) >
cursor.remaining().len() {
+ return Err(ConsumerOffsetsWireError::Truncated);
+ }
let mut entries = Vec::with_capacity(count as usize);
let mut previous: Option<u32> = None;
for _ in 0..count {
@@ -672,7 +718,7 @@ impl std::error::Error for SegmentWalkError {}
/// # Errors
/// [`SegmentWalkError`] on the first invalid byte; nothing is partially
/// trusted.
-pub(crate) fn walk_segment_payload(
+pub(crate) async fn walk_segment_payload(
base_offset: u64,
bytes: &[u8],
) -> Result<(SegmentWalkStats, Vec<u8>), SegmentWalkError> {
@@ -680,7 +726,18 @@ pub(crate) fn walk_segment_payload(
let mut next_offset = base_offset;
let mut stats: Option<SegmentWalkStats> = None;
let mut index_bytes = Vec::new();
+ let mut indexed_position: Option<usize> = None;
+ let mut since_yield = 0usize;
while position < bytes.len() {
+ // The walk re-hashes every message (`decode_batch_slice` verifies
+ // `batch_checksum`), so a multi-GiB artifact would hold the pump --
and
+ // with it consensus ticks and heartbeats for every group on this core
--
+ // for the whole pass. Yield on the same cadence the serving side's
+ // chunked hash uses.
+ if since_yield >= OFFER_HASH_CHUNK_LEN {
+ since_yield = 0;
+ yield_to_reactor().await;
+ }
let batch =
decode_batch_slice(&bytes[position..]).map_err(|source|
SegmentWalkError::Batch {
position: position as u64,
@@ -720,10 +777,24 @@ pub(crate) fn walk_segment_payload(
// client-supplied and would give the installed replica a divergent
// timestamp column (polls and retention keyed differently per node).
let timestamp = header.base_timestamp;
- // One sparse-index entry per batch, pointing at the batch start.
- index_bytes.extend_from_slice(&header.base_offset.to_le_bytes());
- index_bytes.extend_from_slice(×tamp.to_le_bytes());
- index_bytes.extend_from_slice(&(position as u64).to_le_bytes());
+ // STRIDED, not one entry per batch: the origin writes one entry per
+ // flush chunk, and a per-batch index is dense enough that a
transferred
+ // segment never fits the sealed-index residency cap
+ // (`poll_plan::SEALED_INDEX_RESIDENT_MAX_BYTES`), so every sealed poll
+ // would fall back to binary-searching the file with single-entry
preads
+ // -- a slow path `poll_plan` reserves for a
+ // `messages_required_to_save = 1` misconfiguration, which a transfer
+ // would otherwise produce unconditionally. Both consumers do
lower-bound
+ // lookups and recovery walks forward from the last entry by design, so
+ // sparser is correct; the first batch always gets one.
+ let stride_reached = indexed_position
+ .is_none_or(|indexed| position.saturating_sub(indexed) >=
INDEX_STRIDE_BYTES);
+ if stride_reached {
+ indexed_position = Some(position);
+ index_bytes.extend_from_slice(&header.base_offset.to_le_bytes());
+ index_bytes.extend_from_slice(×tamp.to_le_bytes());
+ index_bytes.extend_from_slice(&(position as u64).to_le_bytes());
+ }
stats = Some(stats.map_or(
SegmentWalkStats {
end_offset: batch_end,
@@ -748,6 +819,7 @@ pub(crate) fn walk_segment_payload(
// and `decode_batch_slice` already rejects a body shorter than
// `total_size()`.
position += header.total_size();
+ since_yield += header.total_size();
}
stats.map_or(Err(SegmentWalkError::Empty), |stats| {
Ok((stats, index_bytes))
@@ -1061,8 +1133,10 @@ fn staging_paths(partition_dir: &str, start_offset: u64)
-> (PathBuf, PathBuf) {
/// subdirectory is not one).
///
/// # Errors
-/// The underlying `std::io::Error`; the caller logs and lets the rebuild
-/// re-hydrate whatever remains in place.
+/// The underlying `std::io::Error`. A failure is NOT recoverable by
rebuilding:
+/// the rebuild plants segment 0 with `file_exists = false` and truncates
+/// whatever the failed quarantine left, so callers tombstone the partition and
+/// leave the bytes for an operator.
pub async fn quarantine_segment_files(partition_dir: &str) ->
std::io::Result<String> {
let mut target = None;
for attempt in 0..1000 {
@@ -1079,6 +1153,11 @@ pub async fn quarantine_segment_files(partition_dir:
&str) -> std::io::Result<St
));
};
compio::fs::create_dir_all(&target).await?;
+ // BLOCKING read_dir on the pump: compio-fs 0.12 exposes no async directory
+ // walk, and `spawn_blocking` is not an escape either -- the shard
executors run
+ // `thread_pool_limit(0)`. Bounded by the entry count of ONE partition
directory,
+ // but it is a real stall (and under the write lock at the converge site),
so it
+ // stays recorded rather than hidden.
let entries = std::fs::read_dir(partition_dir)?;
for entry in entries.flatten() {
let path = entry.path();
@@ -1115,7 +1194,17 @@ pub async fn quarantine_segment_files(partition_dir:
&str) -> std::io::Result<St
/// failing either caller for. The CONVERGE sweep is deliberately not this
/// function -- it deletes the live chain as well and must propagate its
/// errors.
+/// Do NOT widen this predicate to the quarantine's three-suffix list if the
two
+/// are ever unified: the keep-lists callers pass hold staging paths only
(purge
+/// passes none), so a wider filter would unlink every live `.log` and `.index`
+/// on the partition -- worst at the reuse scan, which runs at
descriptor-accept
+/// on a serving partition.
pub(crate) async fn sweep_staging_except(partition_dir: &str, keep: &[&Path]) {
+ // BLOCKING read_dir on the pump: compio-fs 0.12 exposes no async directory
+ // walk, and `spawn_blocking` is not an escape either -- the shard
executors run
+ // `thread_pool_limit(0)`. Bounded by the entry count of ONE partition
directory,
+ // but it is a real stall (and under the write lock at the converge site),
so it
+ // stays recorded rather than hidden.
let Ok(entries) = std::fs::read_dir(partition_dir) else {
return;
};
@@ -1362,7 +1451,7 @@ where
.remove(&start_offset);
let mut memo = match memo {
Some(memo) if memo.hashed_len == size => {
- let checksum = memo.checksum;
+ let checksum = memo.hasher.finish();
self.segment_checksum_cache
.borrow_mut()
.insert(start_offset, memo);
@@ -1389,8 +1478,7 @@ where
source,
})?;
memo.hashed_len = size;
- memo.checksum = memo.hasher.finish();
- let checksum = memo.checksum;
+ let checksum = memo.hasher.finish();
self.segment_checksum_cache
.borrow_mut()
.insert(start_offset, memo);
@@ -1403,8 +1491,8 @@ where
/// lives on the free function. `None` for an in-memory partition.
///
/// # Errors
- /// The underlying `std::io::Error`; the caller logs and lets the rebuild
- /// re-hydrate whatever remains in place.
+ /// The underlying `std::io::Error`; see [`quarantine_segment_files`] for
why
+ /// a failure is not something the rebuild can absorb.
pub async fn quarantine_partition_dir(&self) ->
std::io::Result<Option<String>> {
let Some(dir) = self.partition_dir.clone() else {
return Ok(None);
@@ -1493,13 +1581,14 @@ where
// verifies every artifact before decoding: the walk's per-batch
// checksums prove batch bodies, not that these are the bytes the
// manifest promised (length alone is implied by completion).
- if !consensus::verify_state_artifact(entry, &bytes) {
+ if !verify_state_artifact_yielding(entry, &bytes).await {
return Err(SpillError::ManifestChecksum {
frontier: entry.frontier,
});
}
- let (stats, index_bytes) =
- walk_segment_payload(entry.frontier,
&bytes).map_err(SpillError::Walk)?;
+ let (stats, index_bytes) = walk_segment_payload(entry.frontier, &bytes)
+ .await
+ .map_err(SpillError::Walk)?;
let (log_staging, index_staging) = staging_paths(&partition_dir,
entry.frontier);
let index_size = index_bytes.len() as u64;
// Two writes, not a loop: each moves its buffer into compio's
@@ -1522,17 +1611,13 @@ where
path: PathBuf::from(&partition_dir),
source,
})?;
- Ok(StagedSegmentMeta {
- start_offset: entry.frontier,
- end_offset: stats.end_offset,
- size: entry.len,
+ Ok(StagedSegmentMeta::from_walk(
+ entry,
+ stats,
index_size,
- start_timestamp: stats.start_timestamp,
- end_timestamp: stats.end_timestamp,
- max_timestamp: stats.max_timestamp,
log_staging,
index_staging,
- })
+ ))
}
/// Adopt an already-verified staged log without rewriting it: walk the
@@ -1554,8 +1639,9 @@ where
let Some(partition_dir) = self.partition_dir.clone() else {
return Err(SpillError::NoPartitionDir);
};
- let (stats, index_bytes) =
- walk_segment_payload(entry.frontier,
bytes).map_err(SpillError::Walk)?;
+ let (stats, index_bytes) = walk_segment_payload(entry.frontier, bytes)
+ .await
+ .map_err(SpillError::Walk)?;
let (log_staging, index_staging) = staging_paths(&partition_dir,
entry.frontier);
let index_size = index_bytes.len() as u64;
write_staging_file(&index_staging, index_bytes)
@@ -1564,17 +1650,13 @@ where
path: index_staging.clone(),
source,
})?;
- Ok(StagedSegmentMeta {
- start_offset: entry.frontier,
- end_offset: stats.end_offset,
- size: entry.len,
+ Ok(StagedSegmentMeta::from_walk(
+ entry,
+ stats,
index_size,
- start_timestamp: stats.start_timestamp,
- end_timestamp: stats.end_timestamp,
- max_timestamp: stats.max_timestamp,
log_staging,
index_staging,
- })
+ ))
}
/// Scan the partition directory for staging files left by an earlier
@@ -1641,7 +1723,7 @@ where
let Ok(bytes) = compio::fs::read(&log_staging).await else {
continue;
};
- if !consensus::verify_state_artifact(entry, &bytes) {
+ if !verify_state_artifact_yielding(entry, &bytes).await {
continue;
}
if let Ok(meta) = self.adopt_staged_segment(entry, &bytes).await {
@@ -1752,6 +1834,13 @@ where
}
// ---- mutate phase ----
+ // Record the INCOMING frontier before anything destructive: the swap
+ // below unlinks the old chain and makes that durable before the first
+ // staged rename lands, and boot sweeps `.log.staging`, so a crash in
+ // that window would otherwise leave the frontier named by nothing at
+ // all and the replica would re-mint from 0 against a group at N.
+ self.persist_offset_frontier_at(offsets_wire.next_offset)
+ .await;
// The write lock spans the convergence too: a mutate failure leaves
// the segment vectors drained, and a concurrent replicated append
// indexing `segments().len() - 1` on the emptied vec is exactly the
@@ -1784,6 +1873,11 @@ where
.await
.map_err(|source| PartitionInstallError::ConvergeFailed { source
})?;
}
+ // The frontier just moved with nothing durable naming it (an all-GC'd
+ // origin leaves no segment carrying it, and the crash windows inside
+ // the swap leave none either), so record it before returning. Runs for
+ // the converge path too: it seeds the counter from the same artifact.
+ self.persist_offset_frontier().await;
outcome
}
@@ -1969,6 +2063,18 @@ where
path: partition_dir.to_owned(),
source,
})?;
+ // The per-log fsync below lives inside the staged loop, which is
+ // empty on this path, and `install_empty_segment` opens with
+ // `file_exists = false` (no writer-creation fsyncs), so without
+ // this the frontier-bearing dirent is page-cache only: a crash
+ // right after the install boots an empty directory and re-derives
+ // the counter at 0.
+ fsync_dir(partition_dir)
+ .await
+ .map_err(|source| PartitionInstallError::SwapIo {
+ path: partition_dir.to_owned(),
+ source,
+ })?;
} else {
let last = self.log.segments().len() - 1;
let storage = self.log.storages()[last].clone();
@@ -2062,7 +2168,22 @@ where
paths
};
for path in old_consumer_paths.into_iter().chain(old_group_paths) {
- let _ = delete_persisted_offset(&path).await;
+ if let Err(error) = delete_persisted_offset(&path).await {
+ // Not fatal, but not silent either: a stranded file is an id
+ // absent from the NEW table (matching ids get overwritten at
+ // the same path), and boot resurrects it. Sharpest after a
+ // purged origin ships `next_offset = 0`, where the clamp drops
+ // every incoming entry and the whole old table survives while
+ // the install still reports success.
+ tracing::warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.consensus().namespace(),
+ path = %path,
+ %error,
+ "failed to unlink a superseded consumer-offset file during
install"
+ );
+ }
}
self.persisted_offsets.borrow_mut().clear();
self.pending_consumer_offset_commits.clear();
@@ -2212,9 +2333,9 @@ where
// PrepareOk'd before the transfer armed, since a transferring replica
// withholds acks -- are ops consensus still claims and the journal can
// no longer serve. The primary's retransmit dies in the backup gap
- // check and repair will not arm (the install leaves
- // `commit_min == commit_max`), so a DVC from here would advertise an
op
- // this replica cannot walk. The pipeline is cleared in the same
breath:
+ // check, so a DVC from here would advertise an op this replica cannot
+ // walk -- tail repair targets the `(commit_op, commit_max]` gap, not
+ // this one. The pipeline is cleared in the same breath:
// its entries are backed by the same erased journal, and
// `LocalPipeline::push` asserts op sequentiality in release, so a bare
// rewind would turn the silent desync into a shard panic on a replica
@@ -2270,6 +2391,11 @@ where
// in-memory vectors were already drained, so only the directory
// itself knows what needs unlinking.
if let Some(partition_dir) = self.partition_dir.clone() {
+ // BLOCKING read_dir on the pump: compio-fs 0.12 exposes no async
directory
+ // walk, and `spawn_blocking` is not an escape either -- the shard
executors run
+ // `thread_pool_limit(0)`. Bounded by the entry count of ONE
partition directory,
+ // but it is a real stall (and under the write lock at the
converge site), so it
+ // stays recorded rather than hidden.
let swept: Vec<PathBuf> = match std::fs::read_dir(&partition_dir) {
Ok(entries) => entries
.flatten()
@@ -2376,6 +2502,27 @@ impl fmt::Display for SpillError {
impl std::error::Error for SpillError {}
+/// Payload bytes between rebuilt sparse-index entries.
+///
+/// Targets the ORIGIN's density (one entry per flush chunk), not maximum
+/// sparseness: at this stride a maximum-size 1 GiB segment rebuilds ~16k
+/// entries (~384 KiB), inside `poll_plan::SEALED_INDEX_RESIDENT_MAX_BYTES`, so
+/// a transferred segment caches its index like any other instead of taking the
+/// per-poll binary-search fallback.
+const INDEX_STRIDE_BYTES: usize = 64 * 1024;
+
+/// Hand the core back to the reactor mid-CPU-pass.
+///
+/// A zero-duration timer, NOT a bare self-waking yield: this runtime does not
+/// reliably re-poll a task that woke itself from inside its own poll, and a
+/// pump that suspends that way stops driving consensus entirely (the frame
+/// handler never resumes, ticks stop, the node goes quiet until something else
+/// wakes it). Registering with the reactor is what every other yield on these
+/// paths does -- the serving side yields through real file reads.
+async fn yield_to_reactor() {
+ compio::time::sleep(std::time::Duration::ZERO).await;
+}
+
/// Chunk size for the offer build's streaming checksum pass. Large enough
/// that per-chunk overhead is noise, small enough that the pump yields to
/// the reactor many times per segment.
@@ -2437,24 +2584,119 @@ async fn hash_segment_range(
/// The serving side runs this on the pump to answer a single chunk request, so
/// it must not hold the core for a whole-file read plus a non-yielding hash
over
/// up to 2 GiB -- long enough to miss heartbeat and view-change deadlines on
-/// every group this shard owns. `None` means the file is shorter than the
entry
-/// or no longer hashes to it (GC unlinked and recreated it, or the bytes
rotted):
-/// the caller evicts the offer and has the requester restart, which is the
only
-/// way the load-time re-verification this exists for can be acted on.
+/// every group this shard owns.
///
/// The file may legitimately be LONGER than the entry (an active segment that
/// kept appending after the offer was built); the artifact is the prefix.
+///
+/// # Errors
+/// [`SegmentLoadError`], which the caller maps onto the refusal it sends: a
+/// collapsed `Option` here told a requester that a dying disk was a momentary
+/// blip forever, because the refusal it drives is classified by cause.
pub async fn load_verified_segment_artifact(
log_path: &str,
entry: &consensus::StateArtifact,
-) -> Option<Vec<u8>> {
+) -> Result<Vec<u8>, SegmentLoadError> {
let mut hasher = consensus::state_manifest::StateArtifactHasher::new();
#[allow(clippy::cast_possible_truncation)]
let mut bytes = Vec::with_capacity(entry.len as usize);
hash_segment_range(log_path, 0, entry.len, &mut hasher, Some(&mut bytes))
.await
- .ok()?;
- (hasher.finish() == entry.checksum).then_some(bytes)
+ .map_err(SegmentLoadError::classify)?;
+ if hasher.finish() != entry.checksum {
+ return Err(SegmentLoadError::ChecksumMismatch);
+ }
+ Ok(bytes)
+}
+
+/// Why a served segment could not be handed to a requester.
+///
+/// The split is the whole point: a short read is what a concurrent GC
+/// unlink-and-recreate legitimately produces and a checksum mismatch means the
+/// offer is simply stale, but `EIO` / `EACCES` / a failed open is a fault on
+/// THIS node, and telling the requester it was transient hides a dying disk
+/// behind an endless peer rotation.
+#[derive(Debug)]
+pub enum SegmentLoadError {
+ /// The file is gone, shorter than the entry, or otherwise out of step with
+ /// an offer built earlier. Retryable from the requester's side.
+ Stale(std::io::Error),
+ /// The bytes are present but no longer hash to the manifest entry.
+ ChecksumMismatch,
+ /// A local fault: unreadable device, permissions, an open that failed.
+ LocalFault(std::io::Error),
+}
+
+impl SegmentLoadError {
+ fn classify(source: std::io::Error) -> Self {
+ // Only kinds the OS actually named earn the hard verdict:
+ // `hash_segment_range` wraps read failures in `Error::other`, which
+ // erases the kind, and a short read past EOF is the ordinary racing-GC
+ // shape. Everything unclassified is therefore stale (retryable).
+ match source.kind() {
+ std::io::ErrorKind::PermissionDenied => Self::LocalFault(source),
+ _ => Self::Stale(source),
+ }
+ }
+
+ /// Whether the requester should retry without charging a failure.
+ #[must_use]
+ pub const fn transient(&self) -> bool {
+ matches!(self, Self::Stale(_) | Self::ChecksumMismatch)
+ }
+}
+
+impl fmt::Display for SegmentLoadError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::Stale(source) => {
+ write!(f, "served segment no longer matches the offer:
{source}")
+ }
+ Self::ChecksumMismatch => {
+ write!(
+ f,
+ "served segment bytes no longer hash to the manifest entry"
+ )
+ }
+ Self::LocalFault(source) => {
+ write!(f, "served segment is unreadable on this node:
{source}")
+ }
+ }
+ }
+}
+
+impl std::error::Error for SegmentLoadError {}
+
+/// [`consensus::verify_state_artifact`] with reactor yields.
+///
+/// The receiver runs this on the pump for a whole artifact (up to a segment),
+/// and a non-yielding hash of that size makes the node quorum-invisible for
its
+/// duration and starves the same-core segment cleaner.
+async fn verify_state_artifact_yielding(entry: &consensus::StateArtifact,
bytes: &[u8]) -> bool {
+ if bytes.len() as u64 != entry.len {
+ return false;
+ }
+ let mut hasher = consensus::state_manifest::StateArtifactHasher::new();
+ for chunk in bytes.chunks(OFFER_HASH_CHUNK_LEN) {
+ hasher.update(chunk);
+ yield_to_reactor().await;
+ }
+ hasher.finish() == entry.checksum
+}
+
+/// The purge generation an encoded consumer-offsets artifact carries, or `0`
+/// when it cannot be decoded.
+///
+/// Lets the shard refuse an offer built BEFORE a committed purge without
+/// duplicating the wire codec: the install's own generation handling only ever
+/// widens permission, so a stale offer would resurrect purged data with the
+/// local applied generation left at the newer value, which the reconciler's
+/// re-wipe gate then reads as "already applied".
+#[must_use]
+pub fn offered_purge_generation(offsets_bytes: &[u8]) -> u64 {
+ ConsumerOffsetsWire::decode(offsets_bytes)
+ .map(|wire| wire.purge_generation)
+ .unwrap_or_default()
}
/// Stamp over every `SEGMENT_LOG` entry of a manifest, keying
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index cbcd6e9df..fd90b696d 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -955,12 +955,26 @@ evicted_ring_capacity = 4096
# large batches can pin. Must be > 0 and <= "256 MiB".
evicted_ring_bytes_max = "16 MiB"
+# Byte budget for segment payloads a SERVING shard keeps resident to answer
+# state-transfer chunk requests, per shard (process-wide is this times the
shard
+# count). Sized for concurrent pulls: at exactly one maximum segment a single
+# rejoining node thrashes the cache by itself, and every miss re-reads and
+# re-hashes a whole segment to serve one 256 KiB chunk. Must be > 0.
+transfer_served_cache_bytes_max = "4 GB"
+
+# Alloc ceiling for ONE received state-transfer artifact, per shard. The
+# receiver holds it resident through verify, walk and staging write, and up to
+# four transfers run at once. Keep it above the largest legal segment
+# (segment.size plus one max batch) or legal segments are rejected. Must be >
0.
+transfer_artifact_bytes_max = "1088 MB"
+
# Message bus configuration.
# Tunables for the inter-shard / inter-replica internal bus that ships
# consensus traffic between replicas and SDK-client traffic between
# shards. These knobs are consensus-liveness-critical (max_batch gates
# throughput under backpressure). Defaults match
# core::message_bus::config::MessageBusConfig::default().
+
[message_bus]
# Maximum number of BusMessage entries coalesced into a single writev(2)
# call. Hard upper bound: IOV_MAX/2 = 512 on Linux.
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index af72c7d8e..f85d2cfaa 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -26,7 +26,8 @@ use crate::dispatch::{
use crate::http;
use crate::partition_helpers::{
build_partition_fresh, configure_consumer_offsets, ensure_initial_segment,
- open_partition_superblock, restore_partition_view,
validate_namespace_bounds,
+ open_partition_superblock, restore_offset_frontier, restore_partition_view,
+ validate_namespace_bounds,
};
use crate::segment_recovery::{RecoveredSegment, load_persisted_segments};
use crate::server_error::{ServerNgError, ShardJoinFailure,
ShardJoinFailureKind};
@@ -72,8 +73,9 @@ use message_bus::{
};
use metadata::IggyMetadata;
use metadata::MuxStateMachine;
+use metadata::ReplicaIdentity;
use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
-use metadata::impls::recovery::{ReplicaIdentity, recover};
+use metadata::impls::recovery::recover;
use metadata::stm::mux::WithFactory;
use metadata::stm::snapshot::Snapshot;
use metadata::stm::stream::{Partition, Streams};
@@ -1766,15 +1768,32 @@ async fn build_shard_for_thread(
fenced_dir,
"quarantined the refused segment files; they are kept
for inspection"
),
- Err(error) => error!(
- stream_id,
- topic_id,
- partition_id = partition_metadata.id,
- partition_dir,
- %error,
- "failed to quarantine the refused segment files; the
rebuild will \
- re-read them and refuse again"
- ),
+ Err(error) => {
+ // NOT rebuilt: `build_partition_fresh` reaches
+ // `ensure_initial_segment`, which opens segment 0 with
+ // `file_exists = false` and TRUNCATES whatever the
+ // failed quarantine left behind. The likeliest
failures
+ // (suffix cap exhausted, `create_dir_all`) move zero
+ // files, so rebuilding would destroy the oldest
segment
+ // on the first attempt while the higher-offset
survivors
+ // keep refusing every boot -- a loop that never
+ // terminates and eats the chain one segment at a time.
+ // Tombstone instead: the namespace stays
unmaterialised
+ // and unrouted, the reconciler backs off, and an
+ // operator still has every byte.
+ error!(
+ stream_id,
+ topic_id,
+ partition_id = partition_metadata.id,
+ partition_dir,
+ %error,
+ "failed to quarantine the refused segment files;
leaving this \
+ partition tombstoned rather than rebuilding over
them"
+ );
+ partition_stats.zero_out_all();
+ partitions.tombstone(namespace);
+ continue;
+ }
}
// The refused load already folded its segment counts in.
partition_stats.zero_out_all();
@@ -1852,6 +1871,15 @@ async fn build_shard_for_thread(
// Repair pacing is shared by both planes' repair loops, so it is a
// per-shard tunable set once here rather than per consensus group.
shard.set_repair_retry_ticks(repair_retry_ticks(config));
+ shard.set_served_segment_cache_bytes_max(
+ config
+ .partition
+ .transfer_served_cache_bytes_max
+ .as_bytes_u64(),
+ );
+ shard.set_partition_artifact_len_max(
+ config.partition.transfer_artifact_bytes_max.as_bytes_u64(),
+ );
shard.set_repair_chunk_max(config.cluster.repair_chunk_max as u64);
// Bounds a served state-transfer chunk. A frame above the bus ceiling is
// rejected by the RECEIVING transport, which tears the replica connection
@@ -2346,7 +2374,12 @@ async fn load_partition(
partition.dirty_offset.store(counter, Ordering::Relaxed);
partition.should_increment_offset = current_offset.is_some();
partition.stats.set_current_offset(counter);
- let current_offset = counter;
+ // The durable frontier is a LOWER BOUND on top of what the segments
proved:
+ // it is the only carrier left when the segments that named the frontier
are
+ // gone (an all-GC'd origin's install, a crash inside the swap window), and
+ // taking the max means real recovered data always wins.
+ restore_offset_frontier(&mut partition, recovered_state.as_ref());
+ let current_offset = partition.offset.load(Ordering::Acquire);
configure_consumer_offsets(&mut partition, config, namespace,
current_offset)?;
ensure_initial_segment(&mut partition, config, stream_id, topic_id,
partition_id).await?;
diff --git a/core/server-ng/src/partition_helpers.rs
b/core/server-ng/src/partition_helpers.rs
index d722d637e..278a5184d 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -34,7 +34,7 @@ use iggy_common::{
};
use journal::superblock::{PingPongSuperblock, SuperblockContents};
use message_bus::IggyMessageBus;
-use metadata::impls::recovery::{IdentityField, ReplicaIdentity};
+use metadata::{IdentityField, ReplicaIdentity};
use partitions::{IggyIndexWriter, IggyPartition, MessagesWriter, Segment};
use server_common::SegmentStorage;
use server_common::fs_utils::remove_dir_all;
@@ -537,6 +537,46 @@ pub(crate) fn restore_partition_view(
consensus.mark_superblock_durable(state.view, state.log_view);
}
+/// Re-seed a partition's offset counter from the durable frontier, taking the
+/// MAX of what the record holds and what the recovered segments already
proved.
+///
+/// The record is a lower bound, never a completeness claim: it exists because
+/// three paths leave a replica whose counter would otherwise restart at 0
while
+/// the group is at N (a transfer install of an all-GC'd origin, a crash inside
+/// the install's swap window, and the fence-and-rebuild path, which needs no
+/// crash at all). Restarting the counter is not a lag -- replicas re-stamp
+/// `base_offset` from it and recompute `batch_checksum` over the result, so
the
+/// next replicated prepare would persist different bytes here than on every
+/// peer, silently.
+pub(crate) fn restore_offset_frontier(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ recovered: Option<&VsrState>,
+) {
+ let Some(frontier) = recovered
+ .map(|state| state.offset_frontier)
+ .filter(|&f| f > 0)
+ else {
+ return;
+ };
+ let recovered_end = frontier - 1;
+ if partition.should_increment_offset
+ && partition.offset.load(Ordering::Acquire) >= recovered_end
+ {
+ return;
+ }
+ info!(
+ namespace_raw = partition.consensus().namespace(),
+ offset_frontier = frontier,
+ "restored partition offset frontier from its superblock"
+ );
+ partition.offset.store(recovered_end, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(recovered_end, Ordering::Relaxed);
+ partition.should_increment_offset = true;
+ partition.stats.set_current_offset(recovered_end);
+}
+
/// Materialise a brand-new [`IggyPartition`] for a namespace that has no
on-disk state yet.
///
/// Counterpart to bootstrap's `load_partition`, which hydrates from
@@ -675,7 +715,14 @@ pub async fn build_partition_fresh(
"fresh partition must not carry recovered segments"
);
- configure_consumer_offsets(&mut partition, config, namespace, 0)?;
+ // A "fresh" build is also how a FENCED partition comes back (the shard
+ // tombstones it and the reconciler rebuilds through here), and the fence
+ // deliberately leaves the superblock in place, so the recorded frontier is
+ // what stops the rebuild from re-minting offsets the group already used.
+ restore_offset_frontier(&mut partition, recovered_state.as_ref());
+ let current_offset = partition.offset.load(Ordering::Acquire);
+
+ configure_consumer_offsets(&mut partition, config, namespace,
current_offset)?;
ensure_initial_segment(&mut partition, config, stream_id, topic_id,
partition_id).await?;
Ok(partition)
@@ -756,6 +803,7 @@ mod tests {
commit_max: 42,
checkpoint_op: 0,
checkpoint_checksum: 0,
+ offset_frontier: 0,
}
}
diff --git a/core/server-ng/src/segment_recovery.rs
b/core/server-ng/src/segment_recovery.rs
index 297b92afa..edd3d76b1 100644
--- a/core/server-ng/src/segment_recovery.rs
+++ b/core/server-ng/src/segment_recovery.rs
@@ -102,13 +102,15 @@ pub async fn load_persisted_segments(
)
.await?;
- // An index without a single whole entry means any log bytes were torn
- // off mid-write (the crash landed between the message write and the
- // index write). Recover the segment as EMPTY: counting the bytes with
- // `end_offset == start_offset` would fabricate one phantom message for
- // the bootstrap non-empty filters, and appending after the torn bytes
- // would strand undecodable garbage inside the readable range. Zeroed
- // sizes make the next append overwrite the torn bytes instead.
+ // `bounds == None` now means the log holds no whole BATCH either (the
+ // index-less path above already tried walking the log), so there is
+ // nothing to recover: zeroed sizes make the next append overwrite the
+ // torn bytes, where counting them with `end_offset == start_offset`
+ // would fabricate one phantom message for the bootstrap non-empty
+ // filters and strand undecodable garbage inside the readable range.
+ // Note this is NOT tail-only -- a torn index is reachable mid-chain on
+ // the shipped `enforce_fsync = false`, which is why the walk above
+ // exists rather than refusing the partition.
let (start_timestamp, end_timestamp, end_offset,
effective_messages_size) =
if let Some((start_timestamp, end_timestamp, end_offset,
walked_size)) = bounds {
(start_timestamp, end_timestamp, end_offset, walked_size)
@@ -345,6 +347,7 @@ fn file_len(path: &str) -> u64 {
/// `enforce_fsync` there is no ordering barrier between the message write and
/// the index write, and a tail torn mid-flush would otherwise pass while
/// `end_offset` claims offsets whose bytes are incomplete.
+#[allow(clippy::too_many_lines)]
async fn recover_segment_bounds(
index_path: &str,
messages_path: &str,
@@ -432,6 +435,60 @@ async fn recover_segment_bounds(
}
Ok(Some((first.timestamp, end_timestamp, end_offset, position)))
}
+ // No whole index entry, but the log holds bytes: recover the bounds by
+ // WALKING the log from byte 0 instead of declaring the segment empty.
+ //
+ // The index is not the only self-describing copy -- batch headers
carry
+ // their own offsets, timestamps and lengths -- and with the shipped
+ // `enforce_fsync = false` there is no write ordering between a log and
+ // its index, so a torn index is reachable on default config for a
+ // MID-CHAIN segment too, not just the tail. Recovering that as empty
+ // then trips the contiguity guard and refuses the whole partition:
+ // total serve loss (and offset reuse from 0) for a chain whose bytes
+ // are all present. The walk stops at the first header that does not
+ // decode or does not fit, which keeps the torn-tail truncation the
+ // indexed path performs.
+ _ if messages_size > 0 => {
+ let mut position = 0u64;
+ let mut start_timestamp = None;
+ let mut end_offset = start_offset;
+ let mut end_timestamp = 0;
+ while position < messages_size {
+ let Some(header) = read_batch_header(messages_path, position,
messages_size) else {
+ break;
+ };
+ let extent = position.saturating_add(header.total_size() as
u64);
+ if extent > messages_size {
+ break;
+ }
+ if header.message_count > 0 {
+ end_offset = header
+ .base_offset
+ .saturating_add(u64::from(header.message_count) - 1);
+ end_timestamp = header.base_timestamp;
+ start_timestamp.get_or_insert(header.base_timestamp);
+ }
+ position = extent;
+ }
+ let Some(start_timestamp) = start_timestamp else {
+ // Not one whole batch either: the bytes really are unusable,
so
+ // the caller's empty recovery is right after all.
+ return Ok(None);
+ };
+ warn!(
+ stream_id,
+ topic_id,
+ partition_id,
+ start_offset,
+ messages_size,
+ walked_size = position,
+ "sparse index holds no whole entry; recovered segment bounds
by \
+ walking the log instead of discarding it (the index
repopulates \
+ on the next flush, and polls take the index-less fallback
until \
+ then)"
+ );
+ Ok(Some((start_timestamp, end_timestamp, end_offset, position)))
+ }
_ => Ok(None),
}
}
diff --git a/core/server-ng/src/server_error.rs
b/core/server-ng/src/server_error.rs
index 317596069..4345e8f83 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -136,7 +136,7 @@ pub enum ServerNgError {
)]
PartitionSuperblockIdentityMismatch {
dir: PathBuf,
- field: metadata::impls::recovery::IdentityField,
+ field: metadata::IdentityField,
expected: u128,
found: u128,
},
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 428d145ca..ab6129528 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -820,6 +820,15 @@ struct MetadataTransferSession {
peer: u8,
/// Serving peer's applied frontier from the accepted descriptor.
commit_op: u64,
+ /// Snapshot generation of the ACCEPTED descriptor, the key the decode
+ /// budget is charged against.
+ ///
+ /// Recorded at accept because the install-time scan can fail to find it --
+ /// a manifest whose snapshot entry is absent, or a checksum mismatch on an
+ /// earlier artifact aborting the scan -- and an uncharged failure re-armed
+ /// the same peer forever: an unbounded full-manifest re-pull loop. The
+ /// descriptor cannot be accepted without one, so it is always present
here.
+ generation: u64,
/// Empty until the `StateTransferTarget` manifest is accepted, then one
/// entry per offered artifact, pulled in manifest order.
artifacts: Vec<consensus::ArtifactProgress>,
@@ -901,7 +910,8 @@ impl ServedSegmentCache {
/// whole segment to serve one 256 KiB chunk. That is the 4096:1 read
/// amplification this cache exists to prevent, plus an offer eviction per
/// failed re-verify feeding the hard-failure backoff.
- const RESIDENT_BYTES_MAX: u64 = SEGMENT_SIZE_CEILING_BYTES *
Self::CONCURRENT_SERVED_SEGMENTS;
+ const RESIDENT_BYTES_DEFAULT: u64 =
+ SEGMENT_SIZE_CEILING_BYTES * Self::CONCURRENT_SERVED_SEGMENTS;
/// Distinct max-size segments the budget holds at once. Matches the
/// receiver-side in-flight cap, since that is how many distinct segments
one
@@ -940,6 +950,29 @@ impl ServedSegmentCache {
}
}
+ /// Drop every payload cached for `namespace`, crediting their bytes back.
+ ///
+ /// A purge unlinks the segments these payloads copy, and the cache key is
+ /// the manifest checksum over the PRE-purge bytes, so nothing about a hit
+ /// can notice: the serve path answers from the resident copy without
+ /// touching disk, and every served chunk resets the expiry clock, so an
+ /// active puller keeps purged data alive indefinitely.
+ fn evict_namespace(&mut self, namespace: u64) {
+ let stale: Vec<(u64, u64)> = self
+ .entries
+ .keys()
+ .filter(|(entry_namespace, _)| *entry_namespace == namespace)
+ .copied()
+ .collect();
+ for key in stale {
+ if let Some(evicted) = self.entries.remove(&key) {
+ self.resident_bytes = self
+ .resident_bytes
+ .saturating_sub(evicted.payload.len() as u64);
+ }
+ }
+ }
+
fn get(&mut self, namespace: u64, checksum: u64) -> Option<Rc<Vec<u8>>> {
self.use_seq += 1;
let use_seq = self.use_seq;
@@ -950,7 +983,7 @@ impl ServedSegmentCache {
Some(Rc::clone(&cached.payload))
}
- fn insert(&mut self, namespace: u64, checksum: u64, payload: Rc<Vec<u8>>) {
+ fn insert(&mut self, namespace: u64, checksum: u64, payload: Rc<Vec<u8>>,
budget: u64) {
let incoming = payload.len() as u64;
// Credited BEFORE the eviction scan: re-inserting an existing key
frees
// its own slot, and charging that only afterwards evicted neighbours
to
@@ -960,9 +993,7 @@ impl ServedSegmentCache {
.resident_bytes
.saturating_sub(replaced.payload.len() as u64);
}
- while self.resident_bytes.saturating_add(incoming) >
Self::RESIDENT_BYTES_MAX
- && !self.entries.is_empty()
- {
+ while self.resident_bytes.saturating_add(incoming) > budget &&
!self.entries.is_empty() {
let Some((&key, _)) = self
.entries
.iter()
@@ -1057,9 +1088,14 @@ impl<'a> TransferDescriptor<'a> {
/// sends run after the borrow drops.
enum ChunkReply {
Chunk(Message<StateChunkHeader>),
- /// Offer evicted (e.g. the serving process restarted): the requester
- /// gets an unavailable descriptor and restarts its session.
- UnknownOffer,
+ /// Offer evicted (e.g. the serving process restarted, or the segment it
+ /// named can no longer be served): the requester gets an unavailable
+ /// descriptor and restarts its session. `transient` carries whether the
+ /// cause was this node's fault, which is what decides if the requester
+ /// charges a failure.
+ Unavailable {
+ transient: bool,
+ },
}
pub struct IggyShard<B, MJ, S, M, T = (), SB = PingPongSuperblock>
@@ -1220,6 +1256,17 @@ where
/// `[cluster] repair_retry_interval` at bootstrap.
repair_retry_ticks: Cell<u32>,
+ /// Live `[partition] transfer_served_cache_bytes_max`: the byte budget for
+ /// segment payloads this shard keeps resident to serve chunk requests.
+ /// Defaults to [`ServedSegmentCache::RESIDENT_BYTES_DEFAULT`]; server-ng
+ /// overrides it at bootstrap.
+ served_segment_cache_bytes_max: Cell<u64>,
+
+ /// Live `[partition] transfer_artifact_bytes_max`: the alloc ceiling for
one
+ /// RECEIVED artifact. Defaults to `PARTITION_ARTIFACT_LEN_DEFAULT`;
+ /// server-ng overrides it at bootstrap.
+ partition_artifact_len_max: Cell<u64>,
+
/// Live `[message_bus] max_message_size`. Bounds a served state chunk: a
/// frame above this is rejected by the RECEIVING transport, which tears
/// down the whole replica connection. Defaults to a value that leaves
@@ -1346,6 +1393,8 @@ where
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
+ served_segment_cache_bytes_max:
Cell::new(ServedSegmentCache::RESIDENT_BYTES_DEFAULT),
+ partition_artifact_len_max:
Cell::new(Self::PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
@@ -1361,6 +1410,18 @@ where
self.repair_retry_ticks.set(ticks);
}
+ /// Override the serving-side resident payload budget from configuration.
+ /// Called once per shard at bootstrap.
+ pub fn set_served_segment_cache_bytes_max(&self, bytes: u64) {
+ self.served_segment_cache_bytes_max.set(bytes);
+ }
+
+ /// Override the per-artifact receive ceiling from configuration. Called
once
+ /// per shard at bootstrap.
+ pub fn set_partition_artifact_len_max(&self, bytes: u64) {
+ self.partition_artifact_len_max.set(bytes);
+ }
+
/// Override the per-round repair-serving chunk ceiling from configuration.
/// Called once per shard at bootstrap; the simulator and tests keep the
/// compile-time [`REPAIR_CHUNK_MAX`] default.
@@ -1585,6 +1646,8 @@ where
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
+ served_segment_cache_bytes_max:
Cell::new(ServedSegmentCache::RESIDENT_BYTES_DEFAULT),
+ partition_artifact_len_max:
Cell::new(Self::PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
@@ -1903,6 +1966,11 @@ const fn next_transfer_peer(self_id: u8, failed_peer:
u8, replica_count: u8, pri
}
}
+/// Consecutive transient refusals before the re-arm starts logging at `error`,
+/// and the interval it re-logs at afterwards. Sized so a peer that is briefly
+/// behind stays quiet while a partition that never rejoins becomes loud.
+const TRANSFER_REFUSALS_BEFORE_ESCALATION: u32 = 10;
+
/// Exponential re-arm backoff, scaled by the consecutive-failure count and
/// capped at 1024x the base so a long outage settles into a slow poll
/// instead of climbing forever.
@@ -3335,7 +3403,19 @@ where
let config = planes.1.0.config();
// Counted BEFORE the `&mut partition` below exists: the scan takes
// shared borrows of every partition (see `arm_partition_transfer`).
- let transfers_inflight = self.partition_transfers_inflight();
+ // Gated on the arm actually being possible, so a stale or misdirected
+ // frame -- and every StartView for a group that is not awaiting a
+ // transfer, which is all of them during an ordinary view change --
does
+ // not pay a node-wide scan. (A shard-level counter would remove the
scan
+ // entirely, but `IggyPartition::transfer` is `pub` and cleared inside
the
+ // partitions crate, so an externally maintained count would drift;
that
+ // refactor is a prerequisite, not a detail.)
+ let transfers_inflight = if
Self::may_arm_partition_transfer(&planes.1.0, header.namespace)
+ {
+ self.partition_transfers_inflight()
+ } else {
+ 0
+ };
let Some(partition) = self.resolve_partition_target(
&planes.1.0,
header.namespace,
@@ -3680,11 +3760,25 @@ where
let cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
let to_op = header.to_op.min(partition.consensus().commit_max());
- let retained_from =
partition.log.journal().inner.repair_retained_from();
+ // `None` means the journal holds NOTHING, not "nothing was evicted":
+ // the partition journal is memory-only and `clear_all` wipes the
+ // evicted ring with it, so a freshly installed or freshly restarted
+ // peer answers `None` for every op it once had. Reading that as "no
+ // eviction" served a bare `RepairDone`, left the requester's floor at
+ // `None`, and `FloorRefused` -- the ONLY route that arms a partition
+ // state transfer -- never fired: a lagging replica on an idle
+ // partition spun repair forever against a peer-sticky retry. An empty
+ // journal instead reports eviction from the commit frontier, which
+ // refuses the floor into a transfer (the empty window passes the
+ // completeness check) and heals in one round.
+ let retained_from = partition
+ .log
+ .journal()
+ .inner
+ .repair_retained_from()
+ .unwrap_or_else(||
partition.consensus().commit_min().saturating_add(1));
let mut from_op = header.from_op;
- if let Some(retained_from) = retained_from
- && retained_from > from_op
- {
+ if retained_from > from_op {
self.send_repair_range_reply(
cluster,
self_id,
@@ -3956,9 +4050,15 @@ where
}
return;
}
- // Counted BEFORE the `&mut partition` below exists: the scan takes
- // shared borrows of every partition (see `arm_partition_transfer`).
- let transfers_inflight = self.partition_transfers_inflight();
+ // Counted BEFORE the `&mut partition` below exists, and only when an
arm
+ // is possible at all: see the StartView site for why the scan is gated
+ // rather than replaced with a counter.
+ let transfers_inflight = if
Self::may_arm_partition_transfer(&planes.1.0, header.namespace)
+ {
+ self.partition_transfers_inflight()
+ } else {
+ 0
+ };
let config = planes.1.0.config().clone();
let Some(partition) = planes
.1
@@ -4713,6 +4813,9 @@ where
nonce,
peer,
commit_op: 0,
+ // Set when a descriptor is accepted; a session with no accepted
+ // descriptor never reaches the install path that reads it.
+ generation: 0,
artifacts: Vec::new(),
target_accepted: false,
idle_ticks: 0,
@@ -4820,59 +4923,62 @@ where
let served = offers
.get_mut(&(header.namespace, header.replica))
.filter(|served| served.nonce == header.nonce);
- served.map_or(Some(ChunkReply::UnknownOffer), |served| {
- let ServedOffer::Metadata(offer) = &served.offer else {
- return Some(ChunkReply::UnknownOffer);
- };
- // Manifest-index addressing: an index past the offer is a
- // requester bug (or a stale frame) and is dropped below.
- let last_artifact = offer.len().saturating_sub(1);
- let artifact_bytes = offer.payload(header.artifact as usize)?;
- let start = header.offset as usize;
- // A request AT the end of an artifact has nothing left to
serve.
- // Answering it with `Some(&[])` -- which `get(len..len)`
happily
- // returns -- would extend nothing on the receiver, reset both
- // sides' idle counters, and be re-requested at the same offset
- // forever: an unbounded empty-frame ping-pong with the
rejoining
- // replica withholding `PrepareOk` for the life of the process.
- // Reachable when a rebuilt offer is SHORTER than the manifest
the
- // receiver accepted (a client logged out between the two
builds).
- if start >= artifact_bytes.len() {
- return None;
- }
- let end = start
- .saturating_add((header.len as usize).min(chunk_len_max))
- .min(artifact_bytes.len());
- let payload = artifact_bytes.get(start..end)?;
- // Only now that bytes are actually going out: an
out-of-bounds or
- // stale frame must not flip a live offer onto the short
expiry.
- // Tail of the final artifact means the receiver 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;
- }
- // Serving a chunk is the only liveness signal the offer gets;
- // the expiry sweep drops it once these stop arriving. Set here
- // rather than on entry so a request that serves NOTHING cannot
- // keep an abandoned offer alive.
- served.idle_ticks = 0;
- 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);
- Some(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;
- },
- )))
- })
+ served.map_or(
+ Some(ChunkReply::Unavailable { transient: true }),
+ |served| {
+ let ServedOffer::Metadata(offer) = &served.offer else {
+ return Some(ChunkReply::Unavailable { transient: true
});
+ };
+ // Manifest-index addressing: an index past the offer is a
+ // requester bug (or a stale frame) and is dropped below.
+ let last_artifact = offer.len().saturating_sub(1);
+ let artifact_bytes = offer.payload(header.artifact as
usize)?;
+ let start = header.offset as usize;
+ // A request AT the end of an artifact has nothing left to
serve.
+ // Answering it with `Some(&[])` -- which `get(len..len)`
happily
+ // returns -- would extend nothing on the receiver, reset
both
+ // sides' idle counters, and be re-requested at the same
offset
+ // forever: an unbounded empty-frame ping-pong with the
rejoining
+ // replica withholding `PrepareOk` for the life of the
process.
+ // Reachable when a rebuilt offer is SHORTER than the
manifest the
+ // receiver accepted (a client logged out between the two
builds).
+ if start >= artifact_bytes.len() {
+ return None;
+ }
+ let end = start
+ .saturating_add((header.len as
usize).min(chunk_len_max))
+ .min(artifact_bytes.len());
+ let payload = artifact_bytes.get(start..end)?;
+ // Only now that bytes are actually going out: an
out-of-bounds or
+ // stale frame must not flip a live offer onto the short
expiry.
+ // Tail of the final artifact means the receiver 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;
+ }
+ // Serving a chunk is the only liveness signal the offer
gets;
+ // the expiry sweep drops it once these stop arriving. Set
here
+ // rather than on entry so a request that serves NOTHING
cannot
+ // keep an abandoned offer alive.
+ served.idle_ticks = 0;
+ 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);
+ Some(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)) => {
@@ -4881,10 +4987,11 @@ where
.send_to_replica(header.replica,
chunk.into_generic().into_frozen())
.await;
}
- Some(ChunkReply::UnknownOffer) => {
+ Some(ChunkReply::Unavailable { transient }) => {
tracing::info!(
shard = self.id,
requester = header.replica,
+ transient,
"state chunk request for an unknown offer; telling
requester to restart"
);
self.send_state_transfer_target(
@@ -4893,10 +5000,11 @@ where
header.replica,
header.nonce,
header.namespace,
- // Transient for the same reason as the partition arm: an
- // offer that aged out between chunks is not a peer
failure,
- // and the restarted session converges.
- TransferDescriptor::unavailable(true, consensus.view(),
consensus.commit_max()),
+ TransferDescriptor::unavailable(
+ transient,
+ consensus.view(),
+ consensus.commit_max(),
+ ),
)
.await;
}
@@ -5030,6 +5138,12 @@ where
.expect("session checked above");
let peer = session.peer;
let commit_op = session.commit_op;
+ // From the ACCEPTED descriptor, not re-derived from the artifacts: a
+ // scan that aborts before the snapshot entry (unknown kind first, or a
+ // checksum mismatch ahead of it) would leave nothing to charge, and an
+ // uncharged decode failure re-arms the same peer for the same manifest
+ // forever.
+ let generation = session.generation;
// Per-artifact integrity, then pick the pieces this plane installs.
// Unknown kinds are refused rather than skipped: an artifact the
@@ -5037,14 +5151,8 @@ where
// install would otherwise be silently dropped.
let mut snapshot: Option<Vec<u8>> = None;
let mut table: Option<(Vec<u8>, u64)> = None;
- // Captured before the integrity checks so a damaged pull still knows
- // which generation to charge the decode budget against.
- let mut generation: Option<u64> = None;
let mut damaged = false;
for (index, artifact) in session.artifacts.into_iter().enumerate() {
- if artifact.entry.kind ==
consensus::artifact_kind::METADATA_SNAPSHOT {
- generation = Some(artifact.entry.frontier);
- }
let actual = consensus::state_artifact_checksum(&artifact.buf);
if actual != artifact.entry.checksum {
tracing::error!(
@@ -5113,9 +5221,7 @@ where
// failures are charged per snapshot generation instead: a
// generation past its budget is refused at descriptor time until
// the peer checkpoints a new one.
- let exhausted =
- generation.is_some_and(|generation|
self.burn_decode_failure(generation));
- if exhausted {
+ if self.burn_decode_failure(generation) {
tracing::warn!(
shard = self.id,
peer,
@@ -5459,14 +5565,6 @@ where
}
}
- /// Drop serving-side state-transfer offers that stopped being pulled.
- ///
- /// Each offer owns a whole snapshot plus the encoded client table, and the
- /// protocol has no completion frame (a receiver installs and goes quiet),
so
- /// without this a primary that ever served a transfer pins that memory for
- /// the rest of the process. Generous relative to the chunk cadence: a live
- /// puller resets the counter on every chunk it fetches, so only an
abandoned
- /// or finished transfer ages out.
/// Serve one partition `RequestStateTransfer`: build (or re-serve) this
/// group's offer and answer with the descriptor.
#[allow(clippy::future_not_send, clippy::too_many_lines)]
@@ -5509,94 +5607,91 @@ where
served.idle_ticks = 0;
Some(Rc::clone(offer))
});
- if let Some(offer) = cached {
- tracing::debug!(
- shard = self.id,
- namespace_raw = header.namespace,
- requester = header.replica,
- "re-answering a partition state transfer request from the
offer already served"
- );
- let (serving_view, serving_commit_max) =
serving_progress(partition);
- self.send_state_transfer_target(
- cluster,
- self_id,
- header.replica,
- header.nonce,
- header.namespace,
- TransferDescriptor::available(
- &offer.manifest(),
- offer.commit_op,
- serving_view,
- serving_commit_max,
- ),
- )
- .await;
- return;
- }
- match partition.state_transfer_offer(&config).await {
- Ok(offer) => {
- tracing::info!(
- shard = self.id,
- namespace_raw = header.namespace,
- requester = header.replica,
- commit_op = offer.commit_op,
- artifacts = offer.artifact_count(),
- total_len = offer.total_len(),
- "serving partition state transfer"
- );
- let (serving_view, serving_commit_max) =
serving_progress(partition);
- self.send_state_transfer_target(
- cluster,
- self_id,
- header.replica,
- header.nonce,
- header.namespace,
- TransferDescriptor::available(
- &offer.manifest(),
- offer.commit_op,
- serving_view,
- serving_commit_max,
- ),
- )
- .await;
- self.state_transfer_offers.borrow_mut().insert(
- (header.namespace, header.replica),
- ServedStateTransfer {
- nonce: header.nonce,
- offer: ServedOffer::Partition(offer),
- idle_ticks: 0,
- fully_served: false,
- },
- );
- }
- Err(reason) => {
- // The ACTUAL reason: "not the caught-up primary" is routine
- // (the requester re-targets), an unreadable segment is an
- // operator-visible fault on THIS node. The requester cannot
see
- // the reason, only whether it was transient, which is what
keeps
- // a routine refusal from charging its failure count.
- let transient = reason.transient();
- tracing::info!(
+ // One resolve, one send: the three outcomes differ only in the
+ // descriptor they produce, and duplicating the send made it possible
for
+ // them to drift on the progress they advertise.
+ let offer = match cached {
+ Some(offer) => {
+ tracing::debug!(
shard = self.id,
namespace_raw = header.namespace,
requester = header.replica,
- transient,
- %reason,
- "cannot serve partition state transfer; requester falls
back"
+ "re-answering a partition state transfer request from the
offer \
+ already served"
);
- let (serving_view, serving_commit_max) =
serving_progress(partition);
- self.send_state_transfer_target(
- cluster,
- self_id,
- header.replica,
- header.nonce,
- header.namespace,
- TransferDescriptor::unavailable(transient, serving_view,
serving_commit_max),
- )
- .await;
+ Some(offer)
}
- }
+ None => match partition.state_transfer_offer(&config).await {
+ Ok(offer) => {
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ artifacts = offer.artifact_count(),
+ total_len = offer.total_len(),
+ "serving partition state transfer"
+ );
+ self.state_transfer_offers.borrow_mut().insert(
+ (header.namespace, header.replica),
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer: ServedOffer::Partition(Rc::clone(&offer)),
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ Some(offer)
+ }
+ Err(reason) => {
+ // The ACTUAL reason: "not the caught-up primary" is
routine
+ // (the requester re-targets), an unreadable segment is an
+ // operator-visible fault on THIS node. The requester
cannot
+ // see the reason, only whether it was transient, which is
+ // what keeps a routine refusal from charging its failure
+ // count.
+ let transient = reason.transient();
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ requester = header.replica,
+ transient,
+ %reason,
+ "cannot serve partition state transfer; requester
falls back"
+ );
+ let (view, commit_max) = serving_progress(partition);
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ TransferDescriptor::unavailable(transient, view,
commit_max),
+ )
+ .await;
+ return;
+ }
+ },
+ };
+ let Some(offer) = offer else {
+ return;
+ };
+ // Sampled AFTER any build: that build force-flushes and hashes every
+ // un-memoized segment (seconds on a first multi-GiB serve) while
reading
+ // `commit_op` post-flush, so a pre-build sample could advertise a
+ // `commit_max` below the descriptor's own `commit_op` -- which only
+ // makes the receiver's gate refuse, and refusals feed a backoff.
+ let (view, commit_max) = serving_progress(partition);
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ TransferDescriptor::available(&offer.manifest(), offer.commit_op,
view, commit_max),
+ )
+ .await;
}
/// Serve one partition chunk. Segment payloads are loaded from disk on
@@ -5651,10 +5746,14 @@ where
.get_mut(&(header.namespace, header.replica))
.filter(|served| served.nonce == header.nonce);
let Some(served) = served else {
- break 'attempt
ChunkAttempt::Reply(Some(ChunkReply::UnknownOffer));
+ break 'attempt
ChunkAttempt::Reply(Some(ChunkReply::Unavailable {
+ transient: true,
+ }));
};
let ServedOffer::Partition(offer) = &served.offer else {
- break 'attempt
ChunkAttempt::Reply(Some(ChunkReply::UnknownOffer));
+ break 'attempt
ChunkAttempt::Reply(Some(ChunkReply::Unavailable {
+ transient: true,
+ }));
};
let last_artifact = offer.artifact_count().saturating_sub(1);
let artifact = header.artifact as usize;
@@ -5735,25 +5834,35 @@ where
// plus a non-yielding hash over up to 2 GiB holds the core
// long enough to miss the heartbeat and view-change
// deadlines of every group it owns.
- if let Some(bytes) =
partitions::state_transfer::load_verified_segment_artifact(
+ let loaded =
partitions::state_transfer::load_verified_segment_artifact(
&log_path, &entry,
)
- .await
- {
- self.served_segment_cache.borrow_mut().insert(
- header.namespace,
- entry.checksum,
- Rc::new(bytes),
- );
- continue;
- }
+ .await;
+ let reason = match loaded {
+ Ok(bytes) => {
+ self.served_segment_cache.borrow_mut().insert(
+ header.namespace,
+ entry.checksum,
+ Rc::new(bytes),
+ self.served_segment_cache_bytes_max.get(),
+ );
+ continue;
+ }
+ Err(reason) => reason,
+ };
+ // The CAUSE decides what the requester is told: a racing
GC
+ // or a stale offer is transient and costs it nothing,
while
+ // an unreadable device is this node's fault and must
charge,
+ // or a dying disk reads as a momentary blip forever.
+ let transient = reason.transient();
tracing::warn!(
shard = self.id,
namespace_raw = header.namespace,
artifact = header.artifact,
path = %log_path,
- "served segment no longer matches its manifest entry; \
- evicting the offer"
+ transient,
+ %reason,
+ "cannot serve the requested segment; evicting the
offer"
);
self.state_transfer_offers
.borrow_mut()
@@ -5763,7 +5872,7 @@ where
// requester would otherwise be handed the same offer with
// the same dead path, forever.
partition.clear_state_transfer_offer_cache();
- break Some(ChunkReply::UnknownOffer);
+ break Some(ChunkReply::Unavailable { transient });
}
}
};
@@ -5774,11 +5883,12 @@ where
.send_to_replica(header.replica,
chunk.into_generic().into_frozen())
.await;
}
- Some(ChunkReply::UnknownOffer) => {
+ Some(ChunkReply::Unavailable { transient }) => {
tracing::info!(
shard = self.id,
namespace_raw = header.namespace,
requester = header.replica,
+ transient,
"partition chunk request for an unknown offer; telling
requester to restart"
);
self.send_state_transfer_target(
@@ -5787,14 +5897,12 @@ where
header.replica,
header.nonce,
header.namespace,
- // TRANSIENT: routine on a busy primary -- retention GC'd a
- // served segment, or the offer simply aged out of
- // `state_transfer_offers` between two chunks. The retry
- // converges either way (the restarted session reflects the
- // current segment set), so charging the
consecutive-failure
- // count would double the backoff up to 1024x for an event
- // that is not a failure.
- TransferDescriptor::unavailable(true, serving_view,
serving_commit_max),
+ // Usually TRANSIENT -- retention GC'd a served segment, or
+ // the offer aged out between two chunks, and the restarted
+ // session converges -- but a load that failed on a local
+ // fault says so, or a dying disk would read as a momentary
+ // blip forever.
+ TransferDescriptor::unavailable(transient, serving_view,
serving_commit_max),
)
.await;
}
@@ -5819,7 +5927,7 @@ where
/// `mem::take` moves the buffer out of the session, not out of memory, so
it
/// stays resident through verify + walk + staging write, times the
in-flight
/// cap, times the shard count.
- const PARTITION_ARTIFACT_LEN_MAX: u64 =
+ const PARTITION_ARTIFACT_LEN_DEFAULT: u64 =
SEGMENT_SIZE_CEILING_BYTES + SEGMENT_SIZE_OVERSHOOT_BYTES;
/// Sanity cap across a partition manifest. Segment artifacts spill to
@@ -5838,6 +5946,29 @@ where
/// size. Capped-out arms retry via the scheduled re-arm sweep.
const PARTITION_TRANSFERS_INFLIGHT_MAX: usize = 4;
+ /// Whether arming a transfer for `namespace` is even possible right now.
+ ///
+ /// Takes a SHARED borrow and drops it before returning, so a caller may
form
+ /// its `&mut partition` afterwards. The point is to keep the in-flight
scan
+ /// -- which borrows every partition on the shard -- off frames that cannot
+ /// arm anything: a namespace this shard does not own, and the ordinary
case
+ /// of a group that is neither awaiting a transfer nor idle-with-no-re-arm.
+ fn may_arm_partition_transfer(partitions: &IggyPartitions<B, SB>,
namespace_raw: u64) -> bool
+ where
+ B: MessageBus,
+ {
+ partitions
+ .get_by_ns(&IggyNamespace::from_raw(namespace_raw))
+ .is_some_and(|partition| {
+ partition.transfer.is_none()
+ && matches!(
+ partition.consensus().state_transfer_stage(),
+ consensus::StateTransferStage::AwaitingTarget
+ | consensus::StateTransferStage::Idle
+ )
+ })
+ }
+
/// Receiving-side transfers currently in flight on this shard.
///
/// One scan per call, so callers hoist it: with per-partition groups a
@@ -5858,6 +5989,23 @@ where
.count()
}
+ /// Drop every serving-side artifact this shard holds for `namespace`: the
+ /// cached offers and the resident segment payloads behind them.
+ ///
+ /// Called where the partition's bytes stop being the bytes the offers
+ /// describe (a purge). Neither cache can detect that on its own -- offers
+ /// are keyed by `commit_op`, payloads by the manifest checksum over the
old
+ /// bytes -- so a puller mid-transfer would keep receiving purged data and
+ /// keep both expiry clocks reset while doing it.
+ fn drop_served_state_for(&self, namespace: u64) {
+ self.state_transfer_offers
+ .borrow_mut()
+ .retain(|(served_namespace, _), _| *served_namespace != namespace);
+ self.served_segment_cache
+ .borrow_mut()
+ .evict_namespace(namespace);
+ }
+
/// Whether a peer-supplied source replica id names a replica of this
/// cluster.
///
@@ -5919,7 +6067,8 @@ where
namespace_raw = namespace.inner(),
%error,
"failed to quarantine the fenced partition's segment files;
the rebuild \
- re-hydrates whatever they hold and the boot guard refuses a
holed chain"
+ does NOT re-read them -- `build_partition_fresh` plants
segment 0 with \
+ `file_exists = false` and truncates whatever remains"
),
}
self.plane.partitions().tombstone(namespace);
@@ -6042,6 +6191,7 @@ where
where
B: MessageBus + 'static,
T: ShardsTable,
+ M: StreamsFrontend,
{
let header = *msg.header();
// The peer id reaches `next_transfer_peer`'s ring arithmetic through
the
@@ -6138,7 +6288,9 @@ where
// segment cap. An unknown kind is refused here rather than pulled:
// the install cannot represent it anyway.
let kind_capped = entries.iter().all(|entry| match entry.kind {
- consensus::artifact_kind::SEGMENT_LOG => entry.len <=
Self::PARTITION_ARTIFACT_LEN_MAX,
+ consensus::artifact_kind::SEGMENT_LOG => {
+ entry.len <= self.partition_artifact_len_max.get()
+ }
consensus::artifact_kind::CONSUMER_OFFSETS => {
entry.len <= Self::CONSUMER_OFFSETS_ARTIFACT_LEN_MAX
}
@@ -6194,8 +6346,9 @@ where
// session's own promise to bound receiver memory to ONE in-flight
// artifact. Artifacts adopted by the reuse scan below would also be
// reserved and then overwritten with `Staged`, making the retry path's
- // reservation pure waste. `ArtifactProgress::extend_from_chunk`
reserves
- // on the first chunk of its own artifact instead.
+ // reservation pure waste. `append_chunk` reserves the declared length
on
+ // an artifact's FIRST chunk instead, and only ever for the artifact
the
+ // cursor is actually pulling.
session.artifacts = entries
.iter()
.map(|&entry| {
@@ -6223,6 +6376,7 @@ where
where
B: MessageBus + 'static,
T: ShardsTable,
+ M: StreamsFrontend,
{
let header = *msg.header();
let planes = self.plane.inner();
@@ -6263,14 +6417,12 @@ where
where
B: MessageBus + 'static,
T: ShardsTable,
+ M: StreamsFrontend,
{
let planes = self.plane.inner();
let config = planes.1.0.config().clone();
- let Some(partition) = planes
- .1
- .0
- .get_mut_by_ns(&IggyNamespace::from_raw(namespace))
- else {
+ let target_namespace = IggyNamespace::from_raw(namespace);
+ let Some(partition) = planes.1.0.get_mut_by_ns(&target_namespace) else
{
return;
};
// Stage/session desync bail: the probe-exhausted election fallback in
@@ -6365,9 +6517,12 @@ where
let Some(session) = partition.transfer.take() else {
return;
};
- let generation = session.commit_op;
+ // `commit_op`, NOT a "generation": in this file that word means the
+ // committed PURGE generation, and the callee's parameter is
`commit_op`.
+ let commit_op = session.commit_op;
let peer = session.peer;
let mut offsets_bytes: Option<Vec<u8>> = None;
+ let mut offsets_frontier: Option<u64> = None;
let mut damaged = false;
let mut staged = Vec::new();
for artifact in session.artifacts {
@@ -6387,6 +6542,7 @@ where
if offsets_bytes.is_some() {
damaged = true;
} else {
+ offsets_frontier = Some(progress.entry.frontier);
offsets_bytes = Some(progress.buf);
}
}
@@ -6395,6 +6551,23 @@ where
_ => damaged = true,
}
}
+ // Free self-consistency check on a durable input: the builder sets the
+ // descriptor's `commit_op` and the offsets artifact's frontier from
ONE
+ // binding, and `commit_op` goes on to drive `set_commit_floor`,
+ // `set_sequence`, `advance_commit_max` and the reported
+ // `applied_frontier`, while nothing else ever reads that frontier
back.
+ if let Some(frontier) = offsets_frontier
+ && frontier != commit_op
+ {
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace,
+ commit_op,
+ offsets_frontier = frontier,
+ "descriptor commit_op disagrees with its offsets artifact
frontier; refusing the install"
+ );
+ damaged = true;
+ }
let Some(offsets_bytes) = offsets_bytes.filter(|_| !damaged) else {
tracing::warn!(
shard = self.id,
@@ -6405,11 +6578,45 @@ where
.await;
return;
};
+ // A peer that has NOT yet applied a committed purge offers pre-purge
+ // segments under the stale generation. The install's own generation
+ // handling only widens permission (`max`), so it would resurrect the
+ // purged data durably: the local applied value stays at the newer
+ // generation, and the reconciler's `committed > applied` gate never
+ // re-fires. Compared against the METADATA plane's committed value, not
+ // this partition's applied one -- the latter is memory-only and reads 0
+ // after every restart. Routed through the ordinary failure arm, which
+ // rotates the peer; worst case is one wasted pull.
+ let committed_purge_generation = self
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .partition_purge_generation(
+ target_namespace.stream_id(),
+ target_namespace.topic_id(),
+ target_namespace.partition_id(),
+ );
+ let offered_purge_generation =
+
partitions::state_transfer::offered_purge_generation(&offsets_bytes);
+ if offered_purge_generation < committed_purge_generation {
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace,
+ peer,
+ offered_purge_generation,
+ committed_purge_generation,
+ "refusing a partition transfer offer built before a committed
purge; installing it would resurrect purged data"
+ );
+ self.abandon_or_rearm_partition_transfer(partition, peer)
+ .await;
+ return;
+ }
partition
.consensus()
.set_state_transfer_stage(consensus::StateTransferStage::Installing);
let outcome = partition
- .install_state_transfer(&config, generation, staged,
&offsets_bytes)
+ .install_state_transfer(&config, commit_op, staged, &offsets_bytes)
.await;
partition
.consensus()
@@ -6517,6 +6724,23 @@ where
// that spends a minute catching up costs a minute of retries rather
than
// a climb to the 1024x ceiling.
let after_ticks = self.repair_retry_ticks.get();
+ // The flat interval means a partition can sit here for hours without
+ // charging anything, so the ONLY operator signal is this count: it
+ // escalates the log level and feeds a metric, and it never touches the
+ // backoff.
+ let refusals = partition.record_transfer_refusal();
+ self.metrics.record_partition_transfer_refusal();
+ if refusals >= TRANSFER_REFUSALS_BEFORE_ESCALATION
+ && refusals.is_multiple_of(TRANSFER_REFUSALS_BEFORE_ESCALATION)
+ {
+ tracing::error!(
+ shard = self.id,
+ namespace_raw = partition.consensus().namespace(),
+ peer,
+ refusals,
+ "partition state transfer has been refused {refusals} times in
a row; this partition is not rejoining"
+ );
+ }
self.schedule_partition_transfer_rearm(partition, peer, 0, after_ticks)
.await;
}
@@ -6568,6 +6792,16 @@ where
/// Ask for the next missing partition chunk (first unspilled, incomplete
/// artifact in manifest order).
+ ///
+ /// LOCKSTEP by design: one chunk in flight, re-driven per reply, so
transfer
+ /// throughput is `state_chunk_len_max / RTT` -- roughly 26 MB/s at a 10 ms
+ /// link, about 41 s for a 1 GiB segment. `state_chunk_len_max` only clamps
+ /// downward, so no operator knob raises that ceiling; it is worth knowing
+ /// when sizing `segment.size` and retention, since rejoin time scales with
+ /// retained bytes per partition. A small in-flight window would lift it,
but
+ /// it has to grow `[partition] transfer_served_cache_bytes_max` in step
-- that
+ /// budget is sized for exactly the concurrent lockstep pulls the in-flight
+ /// cap allows.
#[allow(clippy::future_not_send)]
async fn request_pending_partition_chunk(&self, namespace: u64)
where
@@ -6609,6 +6843,16 @@ where
}
}
+ /// Drop serving-side state-transfer offers that stopped being pulled.
+ ///
+ /// An offer pins its plane's payload for as long as it lives -- the
metadata
+ /// snapshot plus the encoded client table, or a partition manifest and the
+ /// resident segment payloads behind it -- and the protocol has no
completion
+ /// frame (a receiver installs and goes quiet), so without this a primary
+ /// that ever served a transfer holds that memory for the rest of the
+ /// process. Generous relative to the chunk cadence: a live puller resets
the
+ /// counter on every chunk it fetches, so only an abandoned or finished
+ /// transfer ages out.
fn expire_idle_state_transfer_offers(&self) {
self.served_segment_cache.borrow_mut().expire_idle();
// `max(1)`: the retry interval is operator-configurable, and a zero
would
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index 5b8fa3438..12c11c3bf 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -186,6 +186,7 @@ pub struct ShardMetrics {
partitions_materialised_total: Counter,
partitions_removed_total: Counter,
partitions_reconcile_failures_total: Counter,
+ partition_transfer_refusals_total: Counter,
partition_frames_rejected_stale_total: Counter,
partition_frames_rejected_ahead_total: Counter,
partition_requests_denied_transient_total: Counter,
@@ -218,6 +219,7 @@ impl ShardMetrics {
partitions_materialised_total: Counter::default(),
partitions_removed_total: Counter::default(),
partitions_reconcile_failures_total: Counter::default(),
+ partition_transfer_refusals_total: Counter::default(),
partition_frames_rejected_stale_total: Counter::default(),
partition_frames_rejected_ahead_total: Counter::default(),
partition_requests_denied_transient_total: Counter::default(),
@@ -263,6 +265,18 @@ impl ShardMetrics {
self.partitions_reconcile_failures_total.inc();
}
+ /// Bumped every time a serving peer refuses a partition state transfer.
+ ///
+ /// Transient refusals re-arm on a flat interval and charge no failure
+ /// count -- deliberately, since the alternative routes through a 1024x
+ /// backoff cap that pins a partition for ~17 minutes after the peer has
+ /// already caught up -- which also means a partition stuck rejoining for
+ /// hours produces no signal of its own. This counter plus the escalating
+ /// log level at the refusal site is that signal.
+ pub fn record_partition_transfer_refusal(&self) {
+ self.partition_transfer_refusals_total.inc();
+ }
+
/// Bumped when a parked partition frame is answered instead of served
/// because it was addressed to an incarnation this shard no longer holds
/// (delete + recreate recycled the namespace's slab keys). Serving it
would
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 91e9591fb..ea9c8d206 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -669,12 +669,23 @@ where
&& partition.applied_purge_generation() < generation
{
match partition.purge(&config, generation).await {
- Ok(()) => tracing::debug!(
- shard = self.id,
- namespace_raw = namespace.inner(),
- generation,
- "purge-partition reset partition to empty"
- ),
+ Ok(()) => {
+ // The purge unlinked the very bytes this shard is
+ // serving: the cached offer still advertises the
+ // pre-purge manifest and the payload cache can
+ // answer chunk requests for it without touching
+ // disk, so a puller would install purged data.
Both
+ // are keyed on pre-purge content, so neither can
+ // notice on its own.
+ partition.clear_state_transfer_offer_cache();
+ self.drop_served_state_for(namespace.inner());
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ generation,
+ "purge-partition reset partition to empty"
+ );
+ }
Err(error) => {
// The purge drained every segment before its first
// fallible step, so a failure leaves the partition
