You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/06 12:09:54 UTC
(iggy) 02/02: review changes
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
commit b462ce35693bd010f1bd0373cc868ccd16e7c1c5
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Thu Aug 6 14:09:45 2026 +0200
review changes
---
core/binary_protocol/src/consensus/header.rs | 9 +
core/configs/src/server_ng_config/displays.rs | 20 +-
core/configs/src/server_ng_config/partition.rs | 12 +-
core/consensus/src/vsr_state.rs | 6 +-
core/partitions/src/iggy_partition.rs | 394 +++++++++++++++++++++---
core/partitions/src/journal.rs | 7 +-
core/partitions/src/lib.rs | 2 +-
core/partitions/src/state_transfer.rs | 176 +++++++++--
core/server-ng/config.toml | 22 +-
core/server-ng/src/bootstrap.rs | 7 +-
core/server-ng/src/partition_helpers.rs | 51 +---
core/server-ng/src/partition_reconciler.rs | 26 +-
core/server-ng/src/server_error.rs | 2 +-
core/shard/src/lib.rs | 403 +++++++++++++++++++------
core/shard/src/router.rs | 47 ++-
15 files changed, 943 insertions(+), 241 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index f26e5c71e..a4661cd27 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1495,6 +1495,15 @@ impl ConsensusHeader for StateTransferTargetHeader {
"unavailable_transient must be 0 or 1".to_string(),
));
}
+ // The flag qualifies a refusal, so it is meaningless on an offer. Inert
+ // today (the receiver reads it only inside the `available == 0` arm),
+ // rejected anyway because a self-contradictory descriptor says the
+ // sender is not the build this field was designed for.
+ if self.available == 1 && self.unavailable_transient == 1 {
+ return Err(ConsensusError::InvalidField(
+ "unavailable_transient must be 0 on an available offer".to_string(),
+ ));
+ }
// Unavailable is a bare refusal; a manifest body on it would be
// ambiguous (which offer would the chunks belong to?). An
// `available == 1` body is left unbounded here on purpose: it carries
diff --git a/core/configs/src/server_ng_config/displays.rs b/core/configs/src/server_ng_config/displays.rs
index 2d4047593..ab487129b 100644
--- a/core/configs/src/server_ng_config/displays.rs
+++ b/core/configs/src/server_ng_config/displays.rs
@@ -24,6 +24,7 @@
use super::message_bus::MessageBusConfig;
use super::metadata::MetadataConfig;
+use super::partition::PartitionConfig;
use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig};
use super::server_ng::{ExtraConfig, NamespaceConfig, ServerNgConfig};
use super::tcp::{TcpConfig, TcpSocketConfig, TcpTlsConfig};
@@ -35,7 +36,7 @@ impl Display for ServerNgConfig {
f,
"{{ consumer_group: {}, data_maintenance: {}, extra: {}, message_saver: {}, \
heartbeat: {}, system: {}, quic: {}, tcp: {}, http: {}, telemetry: {}, \
- metadata: {}, message_bus: {} }}",
+ metadata: {}, message_bus: {}, partition: {} }}",
self.consumer_group,
self.data_maintenance,
self.extra,
@@ -48,6 +49,23 @@ impl Display for ServerNgConfig {
self.telemetry,
self.metadata,
self.message_bus,
+ self.partition,
+ )
+ }
+}
+
+impl Display for PartitionConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ prepare_queue_depth: {}, evicted_ring_capacity: {}, \
+ evicted_ring_bytes_max: {}, transfer_served_cache_bytes_max: {}, \
+ transfer_artifact_bytes_max: {} }}",
+ self.prepare_queue_depth,
+ self.evicted_ring_capacity,
+ self.evicted_ring_bytes_max,
+ self.transfer_served_cache_bytes_max,
+ self.transfer_artifact_bytes_max,
)
}
}
diff --git a/core/configs/src/server_ng_config/partition.rs b/core/configs/src/server_ng_config/partition.rs
index 87f8755c9..011e82a9b 100644
--- a/core/configs/src/server_ng_config/partition.rs
+++ b/core/configs/src/server_ng_config/partition.rs
@@ -55,12 +55,16 @@ pub const DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH: usize = 32;
/// sizing endorsement.
pub const MAX_PARTITION_PREPARE_QUEUE_DEPTH: usize = 256;
-/// Mirrors `shard::IggyShard::PARTITION_ARTIFACT_LEN_DEFAULT` (segment ceiling
-/// plus the one whole batch a segment may close past it).
+/// Mirrors the free const `shard::PARTITION_ARTIFACT_LEN_DEFAULT` (segment
+/// ceiling plus the one whole batch a segment may close past it).
pub const DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX: u64 = 1024 * 1024 * 1024 + 64 * 1024 * 1024;
-/// Mirrors `shard::ServedSegmentCache::RESIDENT_BYTES_DEFAULT`.
-pub const DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX: u64 = 2 * 1024 * 1024 * 1024;
+/// Mirrors the free const `shard::SERVED_SEGMENT_CACHE_BYTES_DEFAULT`: room for
+/// two concurrently served segments at the size a SEALED one actually reaches,
+/// which is the artifact ceiling above, not the configured segment target. Sized
+/// off the target instead, two admitted pulls would not both fit and would evict
+/// each other on every chunk.
+pub const DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX: u64 = 2 * DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX;
/// Upper bound on the two state-transfer byte knobs. A typo guard, not a sizing
/// endorsement: both are PER SHARD, so a slipped digit multiplies by the core
diff --git a/core/consensus/src/vsr_state.rs b/core/consensus/src/vsr_state.rs
index 1044dc55e..b9e270168 100644
--- a/core/consensus/src/vsr_state.rs
+++ b/core/consensus/src/vsr_state.rs
@@ -188,7 +188,11 @@ impl fmt::Display for VsrStateError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::WrongLength { expected, actual } => {
- write!(f, "VsrState needs {expected} bytes, got {actual}")
+ write!(
+ f,
+ "VsrState needs {expected} bytes (or {ENCODED_LEN_WITHOUT_FRONTIER}, \
+ the layout before the offset frontier), got {actual}"
+ )
}
Self::LogViewAheadOfView { view, log_view } => write!(
f,
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index bc6d1c246..dad3f70e3 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -72,14 +72,14 @@ use server_common::{
sharding::IggyNamespace,
};
use std::cell::{Cell, RefCell};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::fmt;
use std::hash::Hash;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::Mutex as TokioMutex;
-use tracing::{debug, error, warn};
+use tracing::{debug, warn};
// This struct aliases in terms of the code contained the `LocalPartition from `core/server/src/streaming/partitions/local_partition.rs`.
//
@@ -180,6 +180,16 @@ where
/// terminal policy.
superblock_write_failures: Cell<u64>,
superblock_retry_after_micros: Cell<u64>,
+ /// A committed purge this replica accepted but could not apply, because it
+ /// could not record the frontier reset first. Withholds `PrepareOk` until
+ /// the purge lands: the counter still names the PRE-purge offset space, so
+ /// every op acked meanwhile would be stamped from a `base_offset` the peers
+ /// that already purged do not share.
+ ///
+ /// The superblock persist gate cannot cover this on its own -- it fires on
+ /// `(view, log_view)` changes, and a replica with a stable view and a full
+ /// disk attempts no write, observes no failure, and fences nothing.
+ pub(crate) purge_deferred: bool,
/// The `offset_frontier` the last successful superblock write recorded,
/// seeded at boot from the record that write left behind.
///
@@ -275,6 +285,60 @@ enum Disposition {
},
}
+/// Why a purge did not complete, split by whether it had already mutated.
+///
+/// The two need opposite handling, and conflating them is a data-loss bug:
+/// fencing a partition whose purge failed before it touched anything
+/// quarantines a complete healthy chain while the live counter still names the
+/// pre-purge offset space, and the fence's own frontier write then stamps that
+/// stale counter as durable truth.
+#[derive(Debug)]
+pub enum PurgeError {
+ /// The frontier reset could not be recorded. NOTHING was mutated: the
+ /// segments, the counters and `applied_purge_generation` are all untouched,
+ /// so the reconciler's `committed > applied` gate re-issues this purge on
+ /// its next pass. Retry, do not fence.
+ ///
+ /// Sets [`Self::purge_deferred`], which withholds `PrepareOk` for this
+ /// group until the purge lands, so the replica goes quorum-invisible THERE
+ /// while every other partition on the node keeps serving. Without that
+ /// fence the counter would still name the pre-purge offset space and every
+ /// op this replica acked would be stamped from a `base_offset` its purged
+ /// peers do not share. The superblock persist gate does not cover it: that
+ /// fires on `(view, log_view)` changes, and a stable view attempts no write
+ /// and so observes no failure.
+ ///
+ /// Fencing the SEND rather than the whole partition is the point. The
+ /// alternative was fencing a partition whose chain is still whole, which
+ /// quarantines live data and rebuilds it at the pre-purge frontier.
+ ///
+ /// Carries no cause: the write path reports `bool`, and the underlying
+ /// `ENOSPC` / `EIO` is logged by the superblock writer on the first failure
+ /// and at every power-of-two thereafter.
+ FrontierNotRecorded,
+ /// A step after the drain failed, so the partition holds no serviceable
+ /// segment chain and its next append would panic on `active_segment()`.
+ /// The caller must fence this group for rebuild.
+ Unserviceable(IggyError),
+}
+
+impl fmt::Display for PurgeError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::FrontierNotRecorded => write!(
+ f,
+ "could not record the purge's offset-frontier reset; nothing was mutated"
+ ),
+ Self::Unserviceable(source) => write!(
+ f,
+ "purge left the partition without a serviceable chain: {source}"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for PurgeError {}
+
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct PendingConsumerOffsetCommit {
kind: ConsumerKind,
@@ -376,6 +440,7 @@ where
superblock_lock: LocalGate::new(),
superblock_write_failures: Cell::new(0),
superblock_retry_after_micros: Cell::new(0),
+ purge_deferred: false,
durable_offset_frontier: Cell::new(0),
transfer: None,
transfer_attempts: 0,
@@ -431,11 +496,21 @@ where
}
/// Attach the durable superblock store the boot path opened for this
- /// partition's group. Boot seeds consensus with the recovered
- /// `(view, log_view)` and marks them durable before attaching; from then
- /// on [`Self::persist_superblock_if_needed`] keeps the record current.
- pub fn set_superblock(&mut self, superblock: Rc<SB>) {
+ /// partition's group, along with the record it read back. Boot seeds
+ /// consensus with the recovered `(view, log_view)` and marks them durable
+ /// before attaching; from then on [`Self::persist_superblock_if_needed`]
+ /// keeps the record current.
+ ///
+ /// The record is a PARAMETER rather than a follow-up seeding call because
+ /// the advance direction maxes against its frontier: an attach that left
+ /// that at zero against a record naming N would let the first write after a
+ /// fence lower it, which is the whole defect the field exists to prevent.
+ /// As a separate call it was silently optional, and one of the three attach
+ /// sites dropped it.
+ pub fn set_superblock(&mut self, superblock: Rc<SB>, recovered: Option<&consensus::VsrState>) {
self.superblock = Some(superblock);
+ self.durable_offset_frontier
+ .set(recovered.map_or(0, |state| state.offset_frontier));
}
/// Persist this group's VSR state to its superblock when the view changed
@@ -458,6 +533,7 @@ where
/// withhold the send. The in-memory view stays ahead of the durable one,
/// which a crash safely rolls back, and the next tick retries.
#[allow(clippy::future_not_send)]
+ #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"]
pub async fn persist_superblock_if_needed(&self) -> bool {
let Some(superblock) = self.superblock.as_ref() else {
// No store (in-memory / simulated partitions): nothing can be
@@ -483,7 +559,7 @@ where
// on every 10 ms tick. Back off first, while still reporting `false`
// so the send stays withheld: fail-closed is the point of this gate,
// and the backoff only bounds what the retry costs.
- if self.consensus.clock_realtime_micros() < self.superblock_retry_after_micros.get() {
+ if self.superblock_write_is_backed_off() {
return false;
}
// Re-check needs-persist AFTER acquiring the lock so check and write
@@ -583,6 +659,45 @@ where
}
}
+ /// Re-seed the offset counter from a recovered superblock record, 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.
+ ///
+ /// Lives HERE rather than in the server crate so the boot paths and the
+ /// simulator share one implementation. A copy in the harness was a copy of
+ /// the max rule that had lost the max, in the one place built to catch
+ /// violations of it.
+ pub fn restore_offset_frontier(&mut self, recovered: Option<&consensus::VsrState>) {
+ let Some(frontier) = recovered
+ .map(|state| state.offset_frontier)
+ .filter(|&f| f > 0)
+ else {
+ return;
+ };
+ let recovered_end = frontier - 1;
+ if self.should_increment_offset && self.offset.load(Ordering::Acquire) >= recovered_end {
+ return;
+ }
+ tracing::info!(
+ namespace_raw = self.consensus().namespace(),
+ offset_frontier = frontier,
+ "restored partition offset frontier from its superblock"
+ );
+ self.offset.store(recovered_end, Ordering::Release);
+ self.dirty_offset.store(recovered_end, Ordering::Relaxed);
+ self.should_increment_offset = true;
+ self.stats.set_current_offset(recovered_end);
+ }
+
/// 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]
@@ -606,6 +721,7 @@ where
/// 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)]
+ #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"]
pub async fn persist_offset_frontier(&self) -> bool {
self.persist_offset_frontier_at(self.offset_frontier())
.await
@@ -621,8 +737,9 @@ where
/// just erased, and the following append stamps `base_offset` N where every
/// peer stamps 0.
#[allow(clippy::future_not_send)]
+ #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"]
pub async fn reset_offset_frontier(&self) -> bool {
- self.reset_offset_frontier_to(self.offset_frontier()).await
+ self.reset_offset_frontier_at(self.offset_frontier()).await
}
/// [`Self::reset_offset_frontier`] for a frontier the live counter does not
@@ -637,20 +754,61 @@ where
/// straight back up and leave the pre-purge value on disk across the swap
/// window.
#[allow(clippy::future_not_send)]
- pub async fn reset_offset_frontier_to(&self, frontier: u64) -> bool {
+ #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"]
+ pub async fn reset_offset_frontier_at(&self, frontier: u64) -> bool {
let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
return true;
};
+ if self.superblock_write_is_backed_off() {
+ return false;
+ }
let _superblock_guard = self.superblock_lock.acquire().await;
self.write_superblock_inner(superblock.as_ref(), frontier)
.await
}
- /// Seed the last-written frontier from the record boot read back, so the
- /// first advance maxes against what is actually on disk rather than against
- /// zero. Boot only: every later value comes from a write this replica made.
- pub fn seed_durable_offset_frontier(&self, frontier: u64) {
- self.durable_offset_frontier.set(frontier);
+ /// Record the frontier immediately ahead of an irreversible quarantine,
+ /// BYPASSING the retry backoff.
+ ///
+ /// The gate exists because the other writers' callers became retry loops,
+ /// and skipping a doomed write costs them nothing. This caller is the
+ /// opposite: it writes once and then moves the segments that are the
+ /// record's only corroborating witness into `.fenced.N`, so a skip here is
+ /// not deferred work, it is the last chance gone. A disk that recovered
+ /// inside the backoff window would otherwise leave the rebuild re-seeding
+ /// from a stale record with nothing left to take the max against.
+ ///
+ /// `intended` is the frontier the caller knows the group is at, written
+ /// verbatim; `None` means the live counter is authoritative and the
+ /// advancing form applies.
+ #[allow(clippy::future_not_send)]
+ #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"]
+ pub async fn record_frontier_before_quarantine(&self, intended: Option<u64>) -> bool {
+ let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
+ return true;
+ };
+ let _superblock_guard = self.superblock_lock.acquire().await;
+ match intended {
+ Some(frontier) => {
+ self.write_superblock_inner(superblock.as_ref(), frontier)
+ .await
+ }
+ None => {
+ self.write_superblock(superblock.as_ref(), self.offset_frontier())
+ .await
+ }
+ }
+ }
+
+ /// Whether a recent write failure's backoff window is still open.
+ ///
+ /// The same gate [`Self::persist_superblock_if_needed`] applies before its
+ /// own write, extended to the spelled-value writers because their callers
+ /// became retry loops: a deferred purge is re-issued by the reconciler, and
+ /// without this each pass re-runs a full `atomic_replace` against a disk
+ /// that just refused one, as fast as `ENOSPC` returns.
+ fn superblock_write_is_backed_off(&self) -> bool {
+ self.consensus.clock_realtime_micros() < self.superblock_retry_after_micros.get()
}
/// [`Self::persist_offset_frontier`] for a frontier this replica has not
@@ -664,10 +822,14 @@ where
/// over-claiming is harmless: the convergence that follows a failed install
/// seeds the counter from the same artifact frontier.
#[allow(clippy::future_not_send)]
+ #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"]
pub async fn persist_offset_frontier_at(&self, frontier: u64) -> bool {
let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
return true;
};
+ if self.superblock_write_is_backed_off() {
+ return false;
+ }
let _superblock_guard = self.superblock_lock.acquire().await;
self.write_superblock(superblock.as_ref(), frontier).await
}
@@ -675,6 +837,7 @@ where
/// 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`]).
+ #[must_use = "the bool is the abandon verdict; dropping it disables the stall budget"]
pub const fn burn_transfer_attempt(&mut self) -> bool {
self.transfer_attempts += 1;
self.transfer_attempts > consensus::STATE_TRANSFER_MAX_STALL_RETRIES
@@ -3506,24 +3669,38 @@ where
/// pre-purge frontier at this point.
///
/// # Errors
- /// When the write fails. Refused rather than logged: nothing has been
- /// mutated yet, and a purge that cannot record its reset must not be the
- /// one that erases the data proving the old frontier. The caller fences
- /// this group for rebuild.
+ /// [`PurgeError::FrontierNotRecorded`]. Refused rather than logged: nothing
+ /// has been mutated yet, and a purge that cannot record its reset must not
+ /// be the one that erases the data proving the old frontier. The caller
+ /// RETRIES; it must not fence, since the chain is still whole and the live
+ /// counter still names the pre-purge space.
#[allow(clippy::future_not_send)]
- async fn record_purge_frontier_reset(&self, generation: u64) -> Result<(), IggyError> {
- if self.reset_offset_frontier_to(0).await {
+ async fn record_purge_frontier_reset(&mut self, generation: u64) -> Result<(), PurgeError> {
+ if self.reset_offset_frontier_at(0).await {
+ self.purge_deferred = false;
return Ok(());
}
- error!(
+ self.purge_deferred = true;
+ // The ONLY operator-visible signal for the withhold: `send_prepare_ok`
+ // returns silently, correctly, since it runs per prepare. So this line
+ // has to say that the replica is now out of quorum for this group, or
+ // the symptom reads as a network fault. The consecutive count
+ // correlates it with the superblock writer's own error log, which
+ // carries the `ENOSPC` / `EIO` cause but is rate-limited to
+ // power-of-two failures, while this deferral repeats per reconciler
+ // pass.
+ warn!(
target: "iggy.partitions.diag",
plane = "partitions",
namespace_raw = self.namespace().inner(),
generation,
- "cannot record the purge's offset-frontier reset; refusing to purge so the \
- durable frontier cannot outlive the data it describes"
+ superblock_write_failures = self.superblock_write_failures.get(),
+ "cannot record the purge's offset-frontier reset; deferring the purge so the \
+ durable frontier cannot outlive the data it describes. This replica now \
+ withholds PrepareOk for this partition until the purge lands, so it is \
+ quorum-invisible there; its other partitions are unaffected"
);
- Err(IggyError::Error)
+ Err(PurgeError::FrontierNotRecorded)
}
/// Reset the partition to a single empty segment at offset 0 and clear all
@@ -3537,19 +3714,20 @@ where
/// `PurgeTopic` advances the committed generation and triggers a fresh pass).
///
/// # Errors
- /// If the frontier reset cannot be recorded, which happens before anything
- /// is mutated, or if the replacement segment's log / index file cannot be
- /// created, by which point every segment is already drained. Either way the
- /// caller must FENCE this group (quarantine + retire for the reconciler to
+ /// [`PurgeError::FrontierNotRecorded`] before anything is mutated, which
+ /// the caller RETRIES: the reconciler re-issues the purge while
+ /// `committed > applied`, and fencing a partition that still holds its whole
+ /// chain would quarantine live data behind a counter that still names the
+ /// pre-purge offset space. [`PurgeError::Unserviceable`] once the drain has
+ /// run, which the caller FENCES (quarantine + retire for the reconciler to
/// rebuild), exactly as the state-transfer install's `ConvergeFailed` arm
- /// does: after the drain the next append panics on `active_segment()`, and
- /// before it the partition still holds data the group believes it purged.
+ /// does, or the next append panics on `active_segment()`.
#[allow(clippy::too_many_lines)]
pub async fn purge(
&mut self,
config: &PartitionsConfig,
generation: u64,
- ) -> Result<(), IggyError> {
+ ) -> Result<(), PurgeError> {
let write_lock = self.write_lock.clone();
let _guard = write_lock.lock().await;
@@ -3609,7 +3787,7 @@ where
}
self.reuse_scan_memo.borrow_mut().take();
if let Some(partition_dir) = self.partition_dir.clone() {
- crate::state_transfer::sweep_staging_except(&partition_dir, &[]).await;
+ crate::state_transfer::sweep_staging_except(&partition_dir, &HashSet::new()).await;
}
let start_offset = 0u64;
@@ -3624,8 +3802,11 @@ where
self.dirty_offset.store(start_offset, Ordering::Relaxed);
self.should_increment_offset = false;
- // Recreate a fresh empty segment at offset 0 with real writers.
- self.install_empty_segment(config, start_offset).await?;
+ // Recreate a fresh empty segment at offset 0 with real writers. Every
+ // segment is drained by now, so a failure here is the fence case.
+ self.install_empty_segment(config, start_offset)
+ .await
+ .map_err(PurgeError::Unserviceable)?;
// 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.
@@ -4009,6 +4190,15 @@ where
if !self.persist_superblock_if_needed().await {
return;
}
+ // Same fail-closed shape for a purge this replica accepted but has not
+ // applied: its counter still names the pre-purge offset space, so an ack
+ // now helps commit an op it will stamp differently from every peer that
+ // did apply. The primary's retransmit re-drives the ack once the purge
+ // lands. Local commits still apply -- this fences the SEND, exactly as
+ // the durability gate above does.
+ if self.purge_deferred {
+ return;
+ }
// `VsrAction::RetransmitPrepares` reads from `self.log.journal`.
// Both `SendMessages` (via `append_send_messages_to_journal`) and
// consumer-offset ops (via `apply_replicated_operation`) append
@@ -4320,7 +4510,7 @@ mod tests {
async fn given_advanced_view_when_persist_gate_runs_should_write_vsr_state_once() {
let mut partition = partition_at_view(3, 2);
let store = Rc::new(RecordingSuperblock::default());
- partition.set_superblock(store.clone());
+ partition.set_superblock(store.clone(), None);
assert!(partition.persist_superblock_if_needed().await);
@@ -4361,7 +4551,7 @@ mod tests {
async fn given_record_above_live_counter_when_advancing_should_keep_the_record() {
let mut partition = partition_at_view(1, 1);
let store = Rc::new(RecordingSuperblock::default());
- partition.set_superblock(store.clone());
+ partition.set_superblock(store.clone(), None);
assert!(partition.persist_offset_frontier_at(9_000).await);
assert_eq!(last_recorded_frontier(&store), 9_000);
@@ -4381,6 +4571,37 @@ mod tests {
);
}
+ /// Attaching a store seeds the last-written frontier from the record
+ /// itself, so an advance maxes against what boot read off disk even before
+ /// this replica has written anything. The sibling test reaches that state by
+ /// WRITING first, which cannot catch an attach site that skips the seed.
+ #[compio::test]
+ async fn given_attached_record_when_advancing_should_keep_the_recorded_frontier() {
+ let mut partition = partition_at_view(1, 1);
+ let store = Rc::new(RecordingSuperblock::default());
+ let recovered = consensus::VsrState {
+ cluster: TEST_CLUSTER,
+ replica_id: 0,
+ replica_count: 3,
+ view: 1,
+ log_view: 1,
+ commit_max: 0,
+ checkpoint_op: 0,
+ checkpoint_checksum: 0,
+ offset_frontier: 4_200,
+ };
+ partition.set_superblock(store.clone(), Some(&recovered));
+ assert_eq!(partition.offset_frontier(), 0, "nothing minted locally");
+
+ assert!(partition.persist_offset_frontier().await);
+
+ assert_eq!(
+ last_recorded_frontier(&store),
+ 4_200,
+ "the first write after an attach must not lower the record it was attached to"
+ );
+ }
+
/// The reset direction is the only way down, and it must actually go there:
/// an install under an advancing purge generation records a frontier below
/// the live counter on purpose.
@@ -4388,14 +4609,14 @@ mod tests {
async fn given_reset_below_live_counter_when_written_should_lower_the_record() {
let mut partition = partition_at_view(1, 1);
let store = Rc::new(RecordingSuperblock::default());
- partition.set_superblock(store.clone());
+ partition.set_superblock(store.clone(), None);
partition.offset.store(9_000, Ordering::Release);
partition.should_increment_offset = true;
assert!(partition.persist_offset_frontier().await);
assert_eq!(last_recorded_frontier(&store), 9_001);
- assert!(partition.reset_offset_frontier_to(12).await);
+ assert!(partition.reset_offset_frontier_at(12).await);
assert_eq!(
last_recorded_frontier(&store),
@@ -4412,18 +4633,52 @@ mod tests {
async fn given_failing_store_when_purge_records_its_reset_should_refuse_before_mutating() {
let mut partition = partition_at_view(1, 1);
let store = Rc::new(RecordingSuperblock::default());
- partition.set_superblock(store.clone());
+ partition.set_superblock(store.clone(), None);
partition.offset.store(9_000, Ordering::Release);
partition.should_increment_offset = true;
store.fail_writes.set(true);
- assert!(partition.record_purge_frontier_reset(7).await.is_err());
+ assert!(
+ matches!(
+ partition.record_purge_frontier_reset(7).await,
+ Err(PurgeError::FrontierNotRecorded)
+ ),
+ "the pre-mutation refusal must be distinguishable from a post-drain \
+ failure: the caller retries this one and fences the other"
+ );
+
+ assert!(
+ partition.purge_deferred,
+ "a deferred purge must fence the ack path: the counter still names the \
+ pre-purge offset space, and the view-change persist gate cannot see \
+ this because a stable view attempts no write at all"
+ );
store.fail_writes.set(false);
+ assert!(
+ matches!(
+ partition.record_purge_frontier_reset(7).await,
+ Err(PurgeError::FrontierNotRecorded)
+ ),
+ "the failed write armed a backoff, and the retry must respect it rather \
+ than re-running a full atomic_replace against a disk that just refused one"
+ );
+ assert_eq!(
+ store.attempts.get(),
+ 1,
+ "the backed-off retry must not reach the store at all"
+ );
+
+ // Backoff expiry, without a controllable clock in this fixture.
+ partition.superblock_retry_after_micros.set(0);
partition
.record_purge_frontier_reset(7)
.await
- .expect("a working store records the reset");
+ .expect("a working store records the reset once the backoff elapses");
+ assert!(
+ !partition.purge_deferred,
+ "recording the reset releases the fence"
+ );
assert_eq!(
last_recorded_frontier(&store),
0,
@@ -4456,7 +4711,7 @@ mod tests {
);
let store = Rc::new(RecordingSuperblock::default());
store.fail_writes.set(true);
- partition.set_superblock(store.clone());
+ partition.set_superblock(store.clone(), None);
// The ack path drops an op past the local head, so the head must cover it.
partition.consensus().sequencer().set_sequence(1);
let size = std::mem::size_of::<PrepareHeader>();
@@ -4500,7 +4755,7 @@ mod tests {
let mut partition = partition_at_view(1, 1);
let store = Rc::new(RecordingSuperblock::default());
store.fail_writes.set(true);
- partition.set_superblock(store.clone());
+ partition.set_superblock(store.clone(), None);
assert!(
!partition.persist_superblock_if_needed().await,
@@ -5850,7 +6105,7 @@ mod tests {
groups: Vec::new(),
};
let refused = partition
- .install_state_transfer(&repair_config(), 12, Vec::new(), &behind.encode())
+ .install_state_transfer(&repair_config(), 12, Vec::new(), &behind.encode(), 0)
.await;
assert!(
matches!(
@@ -5867,7 +6122,9 @@ mod tests {
// A purge at the origin is the one legitimate rewind, and the artifact
// carries the generation that proves it: the same offer passes the fence
- // once its generation advances.
+ // once its generation advances past the COMMITTED one the caller reads
+ // off the metadata plane (0 here), not past this replica's memory-only
+ // applied value.
let purged = crate::state_transfer::ConsumerOffsetsWire {
purge_generation: 1,
next_offset: 0,
@@ -5875,7 +6132,7 @@ mod tests {
groups: Vec::new(),
};
let accepted = partition
- .install_state_transfer(&repair_config(), 12, Vec::new(), &purged.encode())
+ .install_state_transfer(&repair_config(), 12, Vec::new(), &purged.encode(), 0)
.await;
assert!(
!matches!(
@@ -5888,6 +6145,51 @@ mod tests {
let _ = std::fs::remove_dir_all(&partition_dir);
}
+ /// The canonical post-restart rejoin: this replica applied a purge before
+ /// the restart, so the metadata plane's COMMITTED generation is 1 while its
+ /// own memory-only `applied_purge_generation` is back at 0. Gated on the
+ /// local field, `offered(1) > applied(0)` reads as an advancing purge and
+ /// disables the rewind refusal -- on the one path it exists to guard.
+ #[compio::test]
+ async fn given_restarted_replica_when_offer_matches_committed_purge_should_refuse_rewind() {
+ let partition_dir = transfer_fence_dir("restart-purge-rewind").await;
+ let mut partition = test_partition();
+ partition.set_partition_dir(partition_dir.clone());
+ partition.should_increment_offset = true;
+ partition.offset.store(99, Ordering::Release);
+ assert_eq!(
+ partition.applied_purge_generation(),
+ 0,
+ "the local generation is memory-only and starts over after a restart"
+ );
+
+ let offer = crate::state_transfer::ConsumerOffsetsWire {
+ purge_generation: 1,
+ next_offset: 50,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ };
+ let refused = partition
+ .install_state_transfer(&repair_config(), 12, Vec::new(), &offer.encode(), 1)
+ .await;
+
+ assert!(
+ matches!(
+ refused,
+ Err(
+ crate::state_transfer::PartitionInstallError::OfferRewindsDurableData {
+ offer_next_offset: 50,
+ local_next_offset: 100,
+ }
+ )
+ ),
+ "an offer that merely matches the committed generation is not a purge \
+ advancing past it, so the rewind fence must hold: got {refused:?}"
+ );
+
+ let _ = std::fs::remove_dir_all(&partition_dir);
+ }
+
/// Primary-by-index at view 0 with nothing committed refuses to serve: an
/// empty group is trivially "caught up", so this gate is the only thing
/// separating a real primary from a phantom whose directory vanished, whose
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index acc4f926a..c3a5b9657 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -696,8 +696,11 @@ where
}),
};
}
- // Dense window, so a bitset beats a `HashSet`: no hashing per op and one
- // allocation of `expected / 8` bytes.
+ // Dense window, so a flat presence vector beats a `HashSet`: no hashing
+ // per op and one contiguous allocation. One BYTE per op rather than one
+ // bit -- `expected` is bounded by `headers.len()`, so the 8x over a real
+ // bitset buys simpler indexing at a size the caller already holds in
+ // headers.
#[allow(clippy::cast_possible_truncation)]
let expected_len = expected as usize;
let mut present = vec![false; expected_len];
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 7b6446af5..75b4c70ca 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -36,7 +36,7 @@ use iggy_common::IggyError;
pub use iggy_index::IggyIndex;
pub use iggy_index_reader::IggyIndexReader;
pub use iggy_index_writer::IggyIndexWriter;
-pub use iggy_partition::IggyPartition;
+pub use iggy_partition::{IggyPartition, PurgeError};
pub use iggy_partitions::IggyPartitions;
pub use journal::{EVICTED_RING_BYTES_MAX, EVICTED_RING_CAPACITY};
pub use messages_writer::MessagesWriter;
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 7ae3c8ca8..833f07aba 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -41,6 +41,7 @@ use journal::superblock::SuperblockStore;
use message_bus::MessageBus;
use server_common::SegmentStorage;
use server_common::send_messages2::decode_batch_slice;
+use std::collections::HashSet;
use std::fmt;
use std::mem::size_of;
use std::path::{Path, PathBuf};
@@ -930,7 +931,14 @@ pub enum PartitionTransferUnavailable {
/// still unhashed. Progress is memoized, so the next request resumes where
/// this one stopped rather than starting the pass again.
OfferBuildInProgress {
+ /// Bytes hashed so far over the chain AS THIS ROUND SEES IT. Carries
+ /// across rounds through the memo rather than resetting per round, but
+ /// retention GC dropping an already-hashed segment lowers it and
+ /// `remaining` together, so it tracks the live chain, not a monotone
+ /// total.
hashed: u64,
+ /// Bytes of it still unhashed. The pair is the only signal that
+ /// separates a converging multi-round build from a stalled one.
remaining: u64,
},
FlushFailed(iggy_common::IggyError),
@@ -982,7 +990,7 @@ impl fmt::Display for PartitionTransferUnavailable {
}
Self::OfferBuildInProgress { hashed, remaining } => write!(
f,
- "offer checksum pass is {hashed} bytes in with {remaining} to go; \
+ "offer checksum pass has hashed {hashed} bytes with {remaining} to go; \
resuming on the next request"
),
Self::FlushFailed(source) => {
@@ -1003,7 +1011,12 @@ impl std::error::Error for PartitionTransferUnavailable {}
/// and the next offset commit blind-writes those files.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PartitionInstallOutcome {
- pub applied_frontier: u64,
+ /// The consensus op the install applied. Named for what it holds: every
+ /// other `frontier` in this module is a MESSAGE OFFSET
+ /// (`VsrState::offset_frontier`, `StateArtifact::frontier`,
+ /// `installed_frontier`), and op-vs-offset confusion is what produced this
+ /// PR's durability defects.
+ pub applied_commit_op: u64,
/// Every transferred offset file was WRITTEN (and the offset
/// directories fsynced, so the old files' unlinks stick). Not a
/// durability claim for the file contents: `persist_offset` fsyncs only
@@ -1070,6 +1083,13 @@ pub enum PartitionInstallError {
/// whose first use kills the whole shard.
ConvergeFailed {
source: iggy_common::IggyError,
+ /// The offer's frontier, carried because the LIVE counter is not it on
+ /// this path: a mutate failure can leave the counter at its pre-install
+ /// value, and under an advancing purge generation that value is above
+ /// the group's. The fence records this instead, or it would stamp the
+ /// stale counter over the reset the install already made and then
+ /// quarantine the segments that would have contradicted it.
+ frontier: u64,
},
}
@@ -1111,9 +1131,10 @@ impl fmt::Display for PartitionInstallError {
Self::SegmentOpen { path, source } => {
write!(f, "re-opening installed segment {path} failed: {source}")
}
- Self::ConvergeFailed { source } => write!(
+ Self::ConvergeFailed { source, frontier } => write!(
f,
- "post-failure convergence failed, the partition must be fenced: {source}"
+ "post-failure convergence failed at frontier {frontier}, \
+ the partition must be fenced: {source}"
),
}
}
@@ -1248,7 +1269,7 @@ pub async fn quarantine_segment_files(partition_dir: &str) -> std::io::Result<St
/// 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]) {
+pub(crate) async fn sweep_staging_except(partition_dir: &str, keep: &HashSet<&Path>) {
let Ok(entries) = segment_dir_entries(partition_dir) else {
return;
};
@@ -1256,7 +1277,7 @@ pub(crate) async fn sweep_staging_except(partition_dir: &str, keep: &[&Path]) {
let is_staging = path
.to_str()
.is_some_and(|path| path.ends_with(STAGING_SUFFIX));
- if is_staging && !keep.contains(&path.as_path()) {
+ if is_staging && !keep.contains(path.as_path()) {
let _ = compio::fs::remove_file(&path).await;
}
}
@@ -1425,18 +1446,27 @@ where
// until the offer completes.
let mut budget = OFFER_HASH_BUDGET_PER_ROUND_BYTES;
let mut segments = Vec::with_capacity(planned.len());
- for (index, (start_offset, size, log_path)) in planned.iter().enumerate() {
+ for (start_offset, size, log_path) in &planned {
let Some(checksum) = self
.segment_checksum(*start_offset, *size, log_path, &mut budget)
.await?
else {
- let remaining = planned[index..]
- .iter()
- .map(|(_, size, _)| *size)
- .sum::<u64>();
+ // CUMULATIVE across rounds, read back off the memo: per-round
+ // figures are constant by construction (a partial round always
+ // spends exactly the budget and always stops inside one
+ // segment), so they render identically on round 1 and round 30
+ // and an operator cannot tell a converging pass from a wedged
+ // one. This is the only window onto a multi-round build.
+ let hashed = self.hashed_prefix_len(&planned);
+ let total = planned.iter().map(|(_, size, _)| *size).sum::<u64>();
+ // The completing round's sweep is skipped on this path, so
+ // prune here too: retention GC can unlink segments across a
+ // long build, and their memos would otherwise accumulate until
+ // some round finally runs the loop to the end.
+ self.retain_segment_checksum_memos(&planned);
return Err(PartitionTransferUnavailable::OfferBuildInProgress {
- hashed: OFFER_HASH_BUDGET_PER_ROUND_BYTES.saturating_sub(budget),
- remaining,
+ hashed,
+ remaining: total.saturating_sub(hashed),
});
};
segments.push(SegmentArtifactSource {
@@ -1480,7 +1510,18 @@ where
.retain(|start_offset, _| live.contains_key(start_offset));
}
+ // Second phantom gate, on the BUILT offer rather than on `commit_max`.
+ // The gate above keys on `commit_max() == 0`, which a replica that lifted
+ // its commit floor through an offsets-only repair window clears while
+ // still holding zero bytes; such a replica passes `is_caught_up_primary`
+ // and would hand a data-holding peer an empty chain at frontier 0,
+ // making it unlink its own. An offer with no segments AND no offset
+ // space is indistinguishable from that phantom, and a group genuinely
+ // in that state has nothing worth transferring anyway.
let offsets_wire = self.offsets_wire_snapshot();
+ if segments.is_empty() && offsets_wire.next_offset == 0 {
+ return Err(PartitionTransferUnavailable::NothingCommitted);
+ }
let offsets_bytes = Rc::new(offsets_wire.encode());
let offsets_entry = consensus::StateArtifact::for_bytes(
artifact_kind::CONSUMER_OFFSETS,
@@ -1496,6 +1537,39 @@ where
Ok(offer)
}
+ /// Bytes of `planned` the memo already covers, clamped per segment to the
+ /// planned length so a memo carrying an active segment's later growth
+ /// cannot report more than this offer will hash.
+ fn hashed_prefix_len(&self, planned: &[(u64, u64, String)]) -> u64 {
+ let memos = self.segment_checksum_cache.borrow();
+ planned
+ .iter()
+ .map(|(start_offset, size, _)| {
+ memos
+ .get(start_offset)
+ .map_or(0, |memo| memo.hashed_len.min(*size))
+ })
+ .sum()
+ }
+
+ /// Drop memo entries whose segment is no longer in `planned`, which is the
+ /// live chain as of this round.
+ ///
+ /// Set-based rather than a scan per entry: this runs on every
+ /// budget-exhausted round, inside the frame body the budget exists to
+ /// bound, and `planned` is capped by `STATE_MANIFEST_ENTRIES_MAX` rather
+ /// than by anything an operator sized, so the quadratic form dominates the
+ /// hashing it was meant to make room for.
+ fn retain_segment_checksum_memos(&self, planned: &[(u64, u64, String)]) {
+ let live: HashSet<u64> = planned
+ .iter()
+ .map(|(start_offset, _, _)| *start_offset)
+ .collect();
+ self.segment_checksum_cache
+ .borrow_mut()
+ .retain(|start_offset, _| live.contains(start_offset));
+ }
+
/// The artifact stamp over the first `size` bytes of a segment file,
/// extending the memoized hasher rather than re-reading what it already
/// covered.
@@ -1823,7 +1897,7 @@ where
matched_paths.clear();
}
// Sweep strays: anything staged that no adopted meta claims.
- let keep: Vec<&Path> = matched_paths.iter().map(PathBuf::as_path).collect();
+ let keep: HashSet<&Path> = matched_paths.iter().map(PathBuf::as_path).collect();
sweep_staging_except(&partition_dir, &keep).await;
*self.reuse_scan_memo.borrow_mut() = Some(ReuseScanMemo {
digest,
@@ -1852,6 +1926,7 @@ where
commit_op: u64,
mut staged: Vec<StagedSegmentMeta>,
offsets_bytes: &[u8],
+ committed_purge_generation: u64,
) -> Result<PartitionInstallOutcome, PartitionInstallError> {
// ---- check phase: nothing below may mutate ----
let Some(partition_dir) = self.partition_dir.clone() else {
@@ -1868,6 +1943,21 @@ where
commit_min,
});
}
+ // The install rewinds the sequencer to `commit_op`, which erases ops
+ // this replica may already have journaled and acked. Bounding it below
+ // by what this replica knows to be COMMITTED keeps the erased window to
+ // ops it does not know are committed -- the checkable form of an
+ // argument the rewind's own comment only asserts. Free on an honest
+ // offer: only a caught-up primary can serve, so its `commit_min`
+ // equals its `commit_max`, and the receiver's descriptor gate already
+ // refused any peer whose `commit_max` was below this one's.
+ let commit_max = self.consensus().commit_max();
+ if commit_op < commit_max {
+ return Err(PartitionInstallError::StaleTransfer {
+ commit_op,
+ commit_min: commit_max,
+ });
+ }
let offsets_wire = ConsumerOffsetsWire::decode(offsets_bytes)?;
// Anti-rewind against the LOCAL OFFSET COUNTER, not the commit
// frontier: the partition journal is memory-only and
@@ -1885,7 +1975,16 @@ where
// replica than on the rest of the group. A purge is the one
// legitimate rewind, and the artifact carries the generation that
// proves one happened.
- let purge_advances = offsets_wire.purge_generation > self.applied_purge_generation;
+ // Against the METADATA plane's committed generation, which the caller
+ // reads off durable state, NOT against `self.applied_purge_generation`:
+ // that one is memory-only and reads 0 after every restart, so a
+ // post-restart rejoin of any ever-purged topic would see
+ // `offered > 0 == applied` and call it an advancing purge. That is the
+ // canonical rejoin, and treating it as a purge disables the
+ // `OfferRewindsDurableData` refusal below -- the one guard standing
+ // between an offer that rewinds this replica's offset space and its
+ // durable data.
+ let purge_advances = offsets_wire.purge_generation > committed_purge_generation;
let local_next_offset = self.offset_frontier();
if !purge_advances && local_next_offset > 0 && offsets_wire.next_offset < local_next_offset
{
@@ -1933,7 +2032,7 @@ where
// above a group that restarted at the offer's, and the next prepare
// would stamp a `base_offset` and `batch_checksum` no peer shares.
let frontier_durable = if purge_advances {
- self.reset_offset_frontier_to(offsets_wire.next_offset)
+ self.reset_offset_frontier_at(offsets_wire.next_offset)
.await
} else {
self.persist_offset_frontier_at(offsets_wire.next_offset)
@@ -1974,7 +2073,10 @@ where
staged_was_empty,
)
.await
- .map_err(|source| PartitionInstallError::ConvergeFailed { source })?;
+ .map_err(|source| PartitionInstallError::ConvergeFailed {
+ source,
+ frontier: offsets_wire.next_offset,
+ })?;
}
// The frontier just moved with nothing durable naming it (an all-GC'd
// origin leaves no segment carrying it, and the crash windows inside
@@ -2017,7 +2119,11 @@ where
// the reuse-scan sweeps too, and boot sweeps ALL of `.staging`
// (`sweep_scratch_files_and_collect_offsets`), so a transfer abandoned
// for good leaks at most until the next restart.
- let keep: Vec<&Path> = staged
+ // A SET, not a list: the sweep tests every staging dirent against this,
+ // and `staged` is peer-supplied up to `STATE_MANIFEST_ENTRIES_MAX`, so a
+ // linear membership test makes the whole sweep quadratic in a number the
+ // requester chooses -- under the partition write lock, with no yields.
+ let keep: HashSet<&Path> = staged
.iter()
.flat_map(|meta| [meta.log_staging.as_path(), meta.index_staging.as_path()])
.collect();
@@ -2109,6 +2215,16 @@ where
path: partition_dir.to_owned(),
source,
})?;
+ // One directory handle for the whole loop. The per-rename fsync STAYS --
+ // each log rename is that segment's commit point and the ordering is the
+ // crash-safety argument -- but re-opening the directory to make each one
+ // is an `open`+`close` per segment for no durability gain.
+ let dir_handle = compio::fs::File::open(partition_dir)
+ .await
+ .map_err(|source| PartitionInstallError::SwapIo {
+ path: partition_dir.to_owned(),
+ source,
+ })?;
for meta in &staged {
let (log_final, _) = final_paths(partition_dir, meta.start_offset);
compio::fs::rename(&meta.log_staging, &log_final)
@@ -2117,7 +2233,8 @@ where
path: log_final.clone(),
source,
})?;
- fsync_dir(partition_dir)
+ dir_handle
+ .sync_all()
.await
.map_err(|source| PartitionInstallError::SwapIo {
path: partition_dir.to_owned(),
@@ -2439,6 +2556,14 @@ where
self.applied_purge_generation = self
.applied_purge_generation
.max(offsets_wire.purge_generation);
+ // Releasing the deferred-purge fence with it. Satisfying the generation
+ // here is what stops the reconciler re-issuing the purge that armed the
+ // fence, so leaving the flag set strands the replica quorum-invisible on
+ // this group for good. The fence's premise is discharged either way:
+ // it exists because the counter still named the pre-purge offset space,
+ // and this install just re-seeded that counter and recorded it durably
+ // before the swap.
+ self.purge_deferred = false;
let consensus = self.consensus();
if commit_op > consensus.commit_min() {
@@ -2465,7 +2590,7 @@ where
self.transfer_offer_cache.borrow_mut().take();
Ok(PartitionInstallOutcome {
- applied_frontier: commit_op,
+ applied_commit_op: commit_op,
offsets_written,
})
}
@@ -2652,9 +2777,14 @@ const OFFER_HASH_CHUNK_LEN: usize = 1 << 20;
/// The pass holds a frame body, and this shard's consensus ticks are a sibling
/// select arm that stays unpolled for its duration, so the budget is really a
/// bound on how long every OTHER group on this core goes without a heartbeat.
-/// 256 MiB is roughly a quarter second at commodity `NVMe` read rates, an order
-/// of magnitude under the shipped `heartbeat_timeout`, and still large enough
-/// that ordinary retention finishes in one round.
+///
+/// A BYTE budget standing in for a time bound, so the margin is storage-class
+/// specific: 256 MiB is roughly a quarter second on commodity `NVMe` against
+/// the shipped 5 s `heartbeat_timeout`, but about 2 s on a throttled cloud
+/// volume at 125 MB/s baseline, which is most of that window. Sized for the
+/// slower case still leaving room, and large enough that ordinary retention
+/// finishes in one round. An elapsed-time clamp would bound it properly on
+/// every storage class.
const OFFER_HASH_BUDGET_PER_ROUND_BYTES: u64 = 256 * 1024 * 1024;
/// Feed bytes `[from, to)` of `path` into `hasher`, read in
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index d29991238..10a529930 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -958,12 +958,20 @@ evicted_ring_bytes_max = "16 MiB"
# Byte budget for segment payloads a SERVING shard keeps resident to answer
# state-transfer chunk requests. PER SHARD, and shard count defaults to core
# count, so the process-wide high-water is this times the core count on top of
-# page cache -- keep that product in mind before raising it. The default holds
-# two maximum-size segments: below one, 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. Running under the budget costs re-reads, not failures.
+# page cache -- keep that product in mind before raising it. The default is a
+# FIXED 2176 MiB: two sealed segments at the SHIPPED system.segment.size of
+# 1 GiB, each of which can close one whole message_bus.max_message_size past
+# its target, which is why it is not 2 GiB. It does not track your segment
+# size. How many groups this shard serves at once IS derived from yours:
+# floor(this / max(partition.transfer_artifact_bytes_max,
+# system.segment.size + 64 MiB)), minimum one. So raising either that knob or
+# system.segment.size without raising this lowers concurrency and can take it
+# to one, serialising rejoins, and nothing at boot warns about it.
+# Below one 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.
+# Running under the budget costs re-reads, not failures.
# Must be > 0 and <= "64 GiB".
-transfer_served_cache_bytes_max = "2 GiB"
+transfer_served_cache_bytes_max = "2176 MiB"
# Alloc ceiling for ONE received state-transfer artifact, per shard. The
# receiver holds it resident through verify, walk and staging write, and up to
@@ -971,7 +979,9 @@ transfer_served_cache_bytes_max = "2 GiB"
# message_bus.max_message_size (a segment may close one whole batch past its
# cap): under that, a legal segment is refused, the whole manifest with it, and
# the partition livelocks re-requesting it from every peer. Boot validates the
-# floor. Must be > 0 and <= "64 GiB".
+# floor. Raising this above the floor for headroom also DIVIDES the serving
+# concurrency derived from transfer_served_cache_bytes_max above, so raise that
+# in step. Must be > 0 and <= "64 GiB".
transfer_artifact_bytes_max = "1088 MiB"
# Message bus configuration.
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 8e4cc506c..57994d9cc 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -26,8 +26,7 @@ use crate::dispatch::{
use crate::http;
use crate::partition_helpers::{
build_partition_fresh, configure_consumer_offsets, ensure_initial_segment,
- open_partition_superblock, restore_offset_frontier, restore_partition_view,
- validate_namespace_bounds,
+ open_partition_superblock, restore_partition_view, validate_namespace_bounds,
};
use crate::segment_recovery::{RecoveredSegment, load_persisted_segments};
use crate::server_error::{ServerNgError, ShardJoinFailure, ShardJoinFailureKind};
@@ -2351,7 +2350,7 @@ async fn load_partition(
})?;
let mut partition = IggyPartition::new(stats.clone(), consensus);
- partition.set_superblock(superblock);
+ partition.set_superblock(superblock, recovered_state.as_ref());
// Recovered partitions honor the same config-surfaced ring ceilings as the
// fresh-create path (build_partition_fresh). Retention is already off for
// single-replica groups, so this only sizes the multi-replica ring.
@@ -2414,7 +2413,7 @@ async fn load_partition(
// 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());
+ partition.restore_offset_frontier(recovered_state.as_ref());
let current_offset = partition.offset.load(Ordering::Acquire);
configure_consumer_offsets(&mut partition, config, namespace, current_offset)?;
diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs
index 2f217004a..020c5155d 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -552,53 +552,6 @@ 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>,
-) {
- // Seeded even when the rest of this function returns early: the advance
- // direction maxes against the last-written value, and leaving it at zero
- // would let the first write after a fence lower the record below what boot
- // just read off disk.
- if let Some(state) = recovered {
- partition.seed_durable_offset_frontier(state.offset_frontier);
- }
- 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
@@ -717,7 +670,7 @@ pub async fn build_partition_fresh(
}
let mut partition = IggyPartition::new(stats, consensus);
- partition.set_superblock(superblock);
+ partition.set_superblock(superblock, recovered_state.as_ref());
// Surface the evicted-ring ceilings from config onto the fresh journal.
// IggyPartition::new has already disabled retention for single-replica
// groups (nobody to serve), so this only sizes the multi-replica ring; the
@@ -751,7 +704,7 @@ pub async fn build_partition_fresh(
// source of truth. Closing it needs the runtime fence to persist the
// frontier before quarantining, and the boot-path chain refusal to carry
// the refused chain's max `end_offset` on its error.
- restore_offset_frontier(&mut partition, recovered_state.as_ref());
+ partition.restore_offset_frontier(recovered_state.as_ref());
let current_offset = partition.offset.load(Ordering::Acquire);
configure_consumer_offsets(&mut partition, config, namespace, current_offset)?;
diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs
index 819465b9a..de8c036fc 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -401,6 +401,13 @@ struct PassCounters {
/// acted on is not answered: aging answers requests, discarding also
/// destroys prepares.
parked_reclaimed: usize,
+ /// Purges staged this pass. Counted so the pass does not arm the
+ /// fast-skip: the pump can DEFER a purge it could not record
+ /// (`PurgeError::FrontierNotRecorded`), which leaves
+ /// `applied_purge_generation` unmoved and bumps no revision, so an armed
+ /// skip would swallow the only re-issue and drop a committed `PurgeTopic`
+ /// on this replica for good.
+ purges_staged: usize,
/// Rebuilds deferred until an in-flight `ConfirmRemove` drains. Counted
/// so the pass does not arm the fast-skip: the pump's drop clears the
/// tombstone and re-wakes us without bumping `Streams::revision`, so an
@@ -418,6 +425,7 @@ impl PassCounters {
+ self.stale
+ self.cg_offsets_purged
+ self.trims_pending
+ + self.purges_staged
+ self.deferred
+ self.parked_reclaimed
}
@@ -473,7 +481,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
reconcile_parked_frames(ctx, &staged, &mut counters);
reconcile_consumer_group_offsets(ctx, &mut counters).await;
reconcile_segment_truncations(ctx, &mut counters);
- reconcile_partition_purges(ctx);
+ reconcile_partition_purges(ctx, &mut counters);
let local_set: AHashSet<IggyNamespace> =
ctx.shard.plane.partitions().namespaces().copied().collect();
@@ -1086,7 +1094,7 @@ fn reconcile_segment_truncations(ctx: &ReconcilerCtx, counters: &mut PassCounter
/// `PurgeTopic` generation is newer than the one the local partition last
/// applied. The pump re-checks the generation before wiping, so a redundant
/// pass (e.g. from an unrelated revision bump) is a no-op.
-fn reconcile_partition_purges(ctx: &ReconcilerCtx) {
+fn reconcile_partition_purges(ctx: &ReconcilerCtx, counters: &mut PassCounters) {
let partitions = ctx.shard.plane.partitions();
let namespaces: Vec<_> = partitions.namespaces().copied().collect();
let streams = ctx.shard.plane.metadata().mux_stm.streams();
@@ -1096,11 +1104,19 @@ fn reconcile_partition_purges(ctx: &ReconcilerCtx) {
namespace.topic_id(),
namespace.partition_id(),
);
- let applied = partitions
- .get_by_ns(&namespace)
- .map_or(0, partitions::IggyPartition::applied_purge_generation);
+ // `namespaces()` is NOT tombstone-filtered while `get_by_ns` is, so an
+ // absent partition would read `applied = 0` and re-stage a purge on
+ // every pass for any ever-purged topic. That was inert while staging
+ // counted as nothing; now that it disarms the fast-skip it would pin
+ // the O(N) scan on forever and enqueue a lifecycle frame per pass that
+ // the pump's tombstone-gated handler silently discards.
+ let Some(partition) = partitions.get_by_ns(&namespace) else {
+ continue;
+ };
+ let applied = partition.applied_purge_generation();
if committed > applied {
ctx.shard.request_purge_partition(namespace, committed);
+ counters.purges_staged += 1;
}
}
}
diff --git a/core/server-ng/src/server_error.rs b/core/server-ng/src/server_error.rs
index 1aa14d62a..49581a190 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -129,7 +129,7 @@ pub enum ServerNgError {
PartitionSuperblockUnverifiable { dir: PathBuf },
#[error(
"partition superblock at {dir} was checksum-clean but did not decode; \
- refusing boot rather than infer a stale view"
+ tombstoning this partition rather than inferring a stale view"
)]
PartitionSuperblockUndecodable {
dir: PathBuf,
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index c4dff2349..803e74de3 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -886,7 +886,7 @@ pub const PARTITION_ARTIFACT_LEN_DEFAULT: u64 =
/// (`[partition] transfer_served_cache_bytes_max`). Pinned like
/// [`PARTITION_ARTIFACT_LEN_DEFAULT`].
pub const SERVED_SEGMENT_CACHE_BYTES_DEFAULT: u64 =
- SEGMENT_SIZE_CEILING_BYTES * CONCURRENT_SERVED_SEGMENTS;
+ PARTITION_ARTIFACT_LEN_DEFAULT * CONCURRENT_SERVED_SEGMENTS;
/// Distinct max-size segments the served-payload budget holds at once.
///
@@ -962,11 +962,15 @@ impl ServedSegmentCache {
/// once the pulls stop).
fn expire_idle(&mut self, idle_sweeps_max: u64) {
self.sweeps += 1;
+ // Strictly BELOW the floor: at `<=` an entry stamped on sweep 0 matches
+ // `0 <= 0` on the very first sweep and is dropped whatever the budget
+ // says, and every other entry loses one sweep of its lifetime. Harmless
+ // in production, but it makes the budget untestable at its boundary.
let floor = self.sweeps.saturating_sub(idle_sweeps_max);
let stale: Vec<(u64, u64)> = self
.entries
.iter()
- .filter(|(_, cached)| cached.last_use_sweep <= floor)
+ .filter(|(_, cached)| cached.last_use_sweep < floor)
.map(|(&key, _)| key)
.collect();
for key in stale {
@@ -1162,6 +1166,15 @@ where
/// `(namespace, requester replica id)`. Bounded by the replica count times
/// the groups this shard serves; replaced per fresh nonce.
state_transfer_offers: RefCell<HashMap<(u64, u8), ServedStateTransfer>>,
+ /// Partition groups with an offer build under way but no offer yet, keyed
+ /// by namespace and carrying ticks since the last request that advanced it.
+ ///
+ /// A build spans rounds (the checksum pass is budgeted per frame body), and
+ /// during those rounds nothing in `state_transfer_offers` names the group,
+ /// so admission control cannot see it without this. Aged out on the same
+ /// clock as an idle offer, since a requester that walked away leaves
+ /// nothing else to release the slot.
+ partition_offer_builds: RefCell<HashMap<u64, u32>>,
/// See [`ServedSegmentCache`].
served_segment_cache: RefCell<ServedSegmentCache>,
@@ -1420,6 +1433,7 @@ where
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
+ partition_offer_builds: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
served_segment_cache_bytes_max: Cell::new(SERVED_SEGMENT_CACHE_BYTES_DEFAULT),
partition_artifact_len_max: Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
@@ -1673,6 +1687,7 @@ where
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
+ partition_offer_builds: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
served_segment_cache_bytes_max: Cell::new(SERVED_SEGMENT_CACHE_BYTES_DEFAULT),
partition_artifact_len_max: Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
@@ -1948,7 +1963,8 @@ where
/// The serving replica's `(view, commit_max)` for a descriptor.
///
/// Sampled per branch, always AFTER any offer build: the build force-flushes and
-/// hashes every un-memoized segment (seconds on a first multi-GiB serve) while
+/// hashes a budgeted slice of the un-memoized segments (a first multi-GiB
+/// serve takes several rounds to complete an offer at all) while
/// reading its `commit_op` post-flush, so a pre-build sample could advertise a
/// `commit_max` below the descriptor's own `commit_op`. Harmless on the receiver
/// (the values are only compared against its own locals) but it makes its gate
@@ -3158,7 +3174,7 @@ where
);
// Recorded view first, exactly as the two boot paths order it: restoring
// after `init` would advertise a view older than the recorded one.
- if let Some(state) = recovered_state {
+ if let Some(state) = recovered_state.as_ref() {
consensus.set_view(state.view);
consensus.set_log_view(state.log_view);
consensus.mark_superblock_durable(state.view, state.log_view);
@@ -3173,8 +3189,14 @@ where
partitions.config().enforce_fsync,
);
if let Some(superblock) = superblock {
- partition.set_superblock(superblock);
- }
+ partition.set_superblock(superblock, recovered_state.as_ref());
+ }
+ // The SAME call the boot paths make, not a copy of it: this restore is
+ // a max against what the segments already proved, and a harness running
+ // a divergent copy of that rule cannot catch a violation of it. Without
+ // the restore at all, a simulator replica rebuilt against a retained
+ // store resumes minting at 0 while its group is at N.
+ partition.restore_offset_frontier(recovered_state.as_ref());
partitions.insert(namespace, partition);
}
@@ -5372,6 +5394,11 @@ where
})
.map(|namespace| async move {
if let Some(partition) = partitions.get_by_ns(&namespace) {
+ // The only dropped durability verdict in the tree: this pre-pass
+ // exists to coalesce the writes, and the per-group loop below re-runs
+ // the same gate on its lock-free fast path and withholds every
+ // view-scoped send when it fails, so the verdict here is redundant
+ // rather than ignored.
let _ = partition.persist_superblock_if_needed().await;
}
})
@@ -5603,16 +5630,51 @@ where
/// segments must be resident at once. A group already being served always
/// passes, so admission cannot revoke a transfer midway.
///
- /// The cap tracks the configured budget rather than a constant, so an
- /// operator who raises `transfer_served_cache_bytes_max` gets the
- /// concurrency it pays for; at least one is always admitted, since refusing
- /// every rejoin is worse than re-reading for a single one.
+ /// BOTH inputs are the configured ones. Dividing by the compile-time
+ /// segment ceiling instead of the deployed `system.segment.size` would make
+ /// the numerator the only thing an operator controls: on a 64 MiB-segment
+ /// deployment the same budget holds sixteen times as many payloads as a cap
+ /// derived from the 1 GiB ceiling would admit, and rejoins serialise for no
+ /// reason.
+ ///
+ /// The divisor is the size a SEALED segment actually reaches, not the
+ /// configured target: rotation fires after the append that crosses it, so a
+ /// sealed segment runs up to one maximum batch past `segment.size`. Dividing
+ /// by the bare target says two payloads fit a two-target budget when they do
+ /// not, and `ServedSegmentCache::insert` then evicts one per chunk -- the
+ /// thrash this cap exists to prevent, reintroduced through the arithmetic.
+ ///
+ /// That size is `partition_artifact_len_max`, which the config validator
+ /// floors at `segment.size` plus the CONFIGURED `message_bus.max_message_size`.
+ /// The compile-time [`SEGMENT_SIZE_OVERSHOOT_BYTES`] only tracks the shipped
+ /// bus cap, so using it would restore the same thrash on any deployment that
+ /// raised that knob: the sealed segment grows with the bus cap while the
+ /// divisor would not. It is kept as a floor for the case where an operator
+ /// sets the artifact ceiling below what a segment can reach.
+ ///
+ /// At least one is always admitted, since refusing every rejoin is worse
+ /// than re-reading for a single one; the quotient rather than the divisor
+ /// carries that clamp, so a zero segment size fails CLOSED at one slot
+ /// instead of disabling admission control.
fn partition_transfer_admission_cap(&self) -> usize {
- let slots = self.served_segment_cache_bytes_max.get() / SEGMENT_SIZE_CEILING_BYTES;
+ let segment_size = self.plane.partitions().config().segment_size.as_bytes_u64();
+ let resident_len = self
+ .partition_artifact_len_max
+ .get()
+ .max(segment_size.saturating_add(SEGMENT_SIZE_OVERSHOOT_BYTES));
+ let slots = self
+ .served_segment_cache_bytes_max
+ .get()
+ .checked_div(resident_len)
+ .unwrap_or(1);
usize::try_from(slots).unwrap_or(usize::MAX).max(1)
}
fn may_serve_another_partition_transfer(&self, namespace: u64) -> bool {
+ let builds = self.partition_offer_builds.borrow();
+ if builds.contains_key(&namespace) {
+ return true;
+ }
let offers = self.state_transfer_offers.borrow();
let mut served: Vec<u64> = offers
.iter()
@@ -5622,6 +5684,12 @@ where
if served.contains(&namespace) {
return true;
}
+ // Builds count too. A multi-round checksum pass holds no offer yet, so
+ // counting only completed offers admitted every requester's whole
+ // in-flight set at once and let each run its own pass: the frame bodies
+ // stay bounded, but the pump carries N budgets per round-cycle and
+ // every other frame, produce included, queues behind them.
+ served.extend(builds.keys().copied());
served.sort_unstable();
served.dedup();
served.len() < self.partition_transfer_admission_cap()
@@ -5714,63 +5782,88 @@ where
.await;
return;
}
- 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"
+ None => {
+ // Claim the admission slot for the whole build, not just for a
+ // completed offer: the checksum pass runs over several rounds
+ // and holds nothing in the offers map meanwhile.
+ self.partition_offer_builds
+ .borrow_mut()
+ .insert(header.namespace, 0);
+ match partition.state_transfer_offer(&config).await {
+ Ok(offer) => {
+ self.partition_offer_builds
+ .borrow_mut()
+ .remove(&header.namespace);
+ 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.
+ //
+ // The slot survives ONLY a budget-exhausted round, which is
+ // a build that will resume; every other refusal abandons
+ // the build and must not keep the group admitted.
+ let building = matches!(
+ reason,
+ partitions::state_transfer::PartitionTransferUnavailable::OfferBuildInProgress { .. }
);
- 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;
+ if !building {
+ self.partition_offer_builds
+ .borrow_mut()
+ .remove(&header.namespace);
+ }
+ 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
+ // Sampled AFTER any build: that build force-flushes and hashes a
+ // budgeted slice of the un-memoized segments (a first multi-GiB serve
+ // spans several rounds before an offer exists) 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.
@@ -6094,6 +6187,10 @@ where
}
fn drop_served_state_for(&self, namespace: u64) {
+ // Including the build slot: the bytes a partial checksum pass covered are
+ // gone with the chain, so the slot behind it is no longer resumable
+ // work and must stop counting against other namespaces' admission.
+ self.partition_offer_builds.borrow_mut().remove(&namespace);
self.state_transfer_offers
.borrow_mut()
.retain(|(served_namespace, _), _| *served_namespace != namespace);
@@ -6140,20 +6237,42 @@ where
/// but the tombstone and the routing row drop SYNCHRONOUSLY here, because
/// the tombstone is the only gate in `get_mut_by_ns` and the queue does not
/// drain until the end of the pump iteration.
+ ///
+ /// `intended_frontier` is the offset frontier the caller knows the group is
+ /// at, for the paths where the LIVE counter is not it. A failed install
+ /// under an advancing purge generation leaves the counter at the pre-purge
+ /// value while the group restarted its offset space lower, and the
+ /// advancing write would stamp that stale counter over the reset the
+ /// install just made, then quarantine the segments that would have
+ /// contradicted it. `None` where the counter is authoritative.
#[allow(clippy::future_not_send)]
async fn fence_partition_for_rebuild(
&self,
namespace: IggyNamespace,
partition: &IggyPartition<B, SB>,
+ intended_frontier: Option<u64>,
) where
B: MessageBus + 'static,
T: ShardsTable,
{
// BEFORE the quarantine: it moves away the segments that are this
// partition's only other witness to the offset frontier, and the
- // rebuild's sole anchor is then the durable record. Advancing form --
- // the live counter is what the rebuild must not fall below.
- partition.persist_offset_frontier().await;
+ // rebuild's sole anchor is then the durable record.
+ // Ungated by the write backoff on purpose: this is a one-shot write
+ // ahead of an irreversible quarantine, not a retry loop, so a skipped
+ // attempt is the last chance gone rather than deferred work.
+ let recorded = partition
+ .record_frontier_before_quarantine(intended_frontier)
+ .await;
+ if !recorded {
+ tracing::error!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ intended_frontier,
+ "could not record the fenced partition's offset frontier before quarantining \
+ its segments; the rebuild will re-seed from whatever the record still holds"
+ );
+ }
match partition.quarantine_partition_dir().await {
Ok(Some(fenced_dir)) => tracing::error!(
shard = self.id,
@@ -6344,8 +6463,22 @@ where
"partition transfer peer cannot serve; backing off before re-arming"
);
if transient {
- self.rearm_partition_transfer_after_refusal(partition, header.replica)
- .await;
+ // The peer that refused is the node that would otherwise serve,
+ // and on the partition arm only a caught-up primary can. Keep
+ // asking it unless it is not the primary this replica knows: a
+ // rotation spends the next round on a backup that can only
+ // refuse, and the serving side's partial offer-build progress
+ // is memoized per node, so that round advances no hashing.
+ let primary = {
+ let consensus = partition.consensus();
+ consensus.primary_index(consensus.view())
+ };
+ self.rearm_partition_transfer_after_refusal(
+ partition,
+ header.replica,
+ header.replica != primary,
+ )
+ .await;
} else {
self.abandon_or_rearm_partition_transfer(partition, header.replica)
.await;
@@ -6360,6 +6493,27 @@ where
// holds; nonce match alone cannot tell the two apart.
let local_view = partition.consensus().view();
let local_commit_max = partition.consensus().commit_max();
+ // `commit_op` past the sender's OWN `commit_max` is self-contradictory:
+ // the offer cannot be built past the frontier its builder had. Nothing
+ // downstream bounds it above -- the install only refuses values BELOW
+ // the local floor, and the offsets-artifact cross-check compares two
+ // numbers the same peer chose -- so without this a peer offering
+ // `commit_op = u64::MAX` drives this replica's commit floor, sequencer
+ // and `commit_max` there and it reports itself fully committed.
+ if header.commit_op > header.commit_max {
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ peer = header.replica,
+ serving_commit_op = header.commit_op,
+ serving_commit_max = header.commit_max,
+ "refusing a partition transfer offer whose commit_op exceeds the sender's \
+ own commit frontier"
+ );
+ self.abandon_or_rearm_partition_transfer(partition, header.replica)
+ .await;
+ return;
+ }
if header.view < local_view || header.commit_max < local_commit_max {
tracing::warn!(
shard = self.id,
@@ -6371,7 +6525,9 @@ where
local_commit_max,
"refusing a partition transfer offer from a replica behind this one"
);
- self.rearm_partition_transfer_after_refusal(partition, header.replica)
+ // ALWAYS rotate: this refusal is evidence about the peer, not about
+ // its timing, so re-asking it is the one thing that cannot help.
+ self.rearm_partition_transfer_after_refusal(partition, header.replica, true)
.await;
return;
}
@@ -6503,7 +6659,15 @@ where
let Some(session) = partition.transfer.as_mut() else {
return;
};
- if session.nonce != header.nonce || !session.target_accepted {
+ // `session.peer` too, not the nonce alone. The other three
+ // partition-transfer handlers all validate their sender; this one
+ // authenticated payload bytes by a 128-bit capability only, which
+ // is thin but real once a peer has seen one frame -- a rotated-away
+ // peer still holds the nonce until the session is re-minted.
+ if session.nonce != header.nonce
+ || session.peer != header.replica
+ || !session.target_accepted
+ {
return;
}
let payload = &msg.as_slice()[size_of::<StateChunkHeader>()..header.size as usize];
@@ -6667,7 +6831,7 @@ where
// 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.
+ // `applied_commit_op`, while nothing else ever reads that frontier back.
if let Some(frontier) = offsets_frontier
&& frontier != commit_op
{
@@ -6728,7 +6892,13 @@ where
.consensus()
.set_state_transfer_stage(consensus::StateTransferStage::Installing);
let outcome = partition
- .install_state_transfer(&config, commit_op, staged, &offsets_bytes)
+ .install_state_transfer(
+ &config,
+ commit_op,
+ staged,
+ &offsets_bytes,
+ committed_purge_generation,
+ )
.await;
partition
.consensus()
@@ -6742,7 +6912,7 @@ where
tracing::info!(
shard = self.id,
namespace_raw = namespace,
- applied_frontier = outcome.applied_frontier,
+ applied_commit_op = outcome.applied_commit_op,
"partition state transfer installed; handing tail to journal repair"
);
} else {
@@ -6752,7 +6922,7 @@ where
tracing::warn!(
shard = self.id,
namespace_raw = namespace,
- applied_frontier = outcome.applied_frontier,
+ applied_commit_op = outcome.applied_commit_op,
"partition state transfer landed WITHOUT fully written consumer \
offsets; the next offset commit rewrites the files"
);
@@ -6761,7 +6931,10 @@ where
self.maybe_request_partition_repair(partition, peer).await;
}
Err(
- error @ partitions::state_transfer::PartitionInstallError::ConvergeFailed { .. },
+ error @ partitions::state_transfer::PartitionInstallError::ConvergeFailed {
+ frontier,
+ ..
+ },
) => {
// The partition holds no serviceable segment chain and its
// next append or poll would panic the shard. Fence exactly
@@ -6777,8 +6950,17 @@ where
%error,
"partition unserviceable after failed install; fencing it for rebuild"
);
- self.fence_partition_for_rebuild(IggyNamespace::from_raw(namespace), partition)
- .await;
+ // Served state first, as the purge fence does: the quarantine
+ // below moves the chain those offers and cached payloads
+ // describe into `.fenced.N`, and a requester holding one would
+ // otherwise pull bytes that no longer exist.
+ self.drop_partition_transfer_state(IggyNamespace::from_raw(namespace), partition);
+ self.fence_partition_for_rebuild(
+ IggyNamespace::from_raw(namespace),
+ partition,
+ Some(frontier),
+ )
+ .await;
}
Err(error) => {
tracing::error!(
@@ -6811,7 +6993,7 @@ where
{
let failures = partition.record_transfer_failure();
let after_ticks = transfer_rearm_backoff(self.repair_retry_ticks.get(), failures);
- self.schedule_partition_transfer_rearm(partition, peer, failures, after_ticks)
+ self.schedule_partition_transfer_rearm(partition, peer, failures, after_ticks, true)
.await;
}
@@ -6824,11 +7006,21 @@ where
/// ceilings: nothing else recovers the partition meanwhile, since repair
/// keeps hitting the refused floor and will not arm while a re-arm is
/// pending.
+ ///
+ /// `rotate` belongs to the CALLER because the two refusal sites mean
+ /// opposite things by it. A peer saying "not right now" is the node that
+ /// would otherwise serve, so staying on it is right. This replica refusing
+ /// a descriptor from a peer that knows LESS than it does is the one case
+ /// where the peer is provably the wrong one, and rotating is the whole
+ /// remedy: a restarted primary comes back at `commit_max = 0` (the
+ /// partition journal is memory-only), so a rejoining backup would otherwise
+ /// pin itself to it at a flat interval until the group's next election.
#[allow(clippy::future_not_send)]
async fn rearm_partition_transfer_after_refusal(
&self,
partition: &mut IggyPartition<B, SB>,
peer: u8,
+ rotate: bool,
) where
B: MessageBus,
{
@@ -6845,20 +7037,33 @@ where
if refusals >= TRANSFER_REFUSALS_BEFORE_ESCALATION
&& refusals.is_multiple_of(TRANSFER_REFUSALS_BEFORE_ESCALATION)
{
- tracing::error!(
+ // Deliberately not phrased as "not rejoining": a serving primary
+ // building a large offer refuses one round per budget slice, so a
+ // healthy multi-GiB rejoin reaches this count while progressing
+ // normally. The descriptor carries no reason code, so this side
+ // cannot tell the two apart; the serving node's own logs can.
+ tracing::warn!(
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"
+ "partition state transfer has been refused {refusals} times in a row; the peer \
+ may be building a large offer or rate-limiting concurrent transfers, or it may \
+ be unable to serve at all -- check its logs before intervening"
);
}
- self.schedule_partition_transfer_rearm(partition, peer, 0, after_ticks)
+ self.schedule_partition_transfer_rearm(partition, peer, 0, after_ticks, rotate)
.await;
}
- /// Drop the session, rotate the peer, and schedule the re-arm; shared by the
- /// charged and uncharged paths.
+ /// Drop the session, pick the next peer, and schedule the re-arm; shared by
+ /// the charged and uncharged paths.
+ ///
+ /// `rotate` is false where the refusing peer is the only one that could
+ /// have served: only a caught-up primary passes `is_caught_up_primary`, so
+ /// rotating off it asks a backup that can answer nothing but another
+ /// refusal, and the serving side's partial offer-build progress is memoized
+ /// PER NODE, so the round spent on the backup also advances no hashing.
#[allow(clippy::future_not_send)]
async fn schedule_partition_transfer_rearm(
&self,
@@ -6866,6 +7071,7 @@ where
peer: u8,
failures: u32,
after_ticks: u32,
+ rotate: bool,
) where
B: MessageBus,
{
@@ -6874,12 +7080,16 @@ where
if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
}
- let next_peer = next_transfer_peer(
- consensus.replica(),
- peer,
- consensus.replica_count(),
- consensus.primary_index(consensus.view()),
- );
+ let next_peer = if rotate {
+ next_transfer_peer(
+ consensus.replica(),
+ peer,
+ consensus.replica_count(),
+ consensus.primary_index(consensus.view()),
+ )
+ } else {
+ peer
+ };
tracing::info!(
shard = self.id,
namespace_raw = partition.consensus().namespace(),
@@ -6979,6 +7189,23 @@ where
let retry_ticks = self.repair_retry_ticks.get().max(1);
let idle_expiry_ticks = retry_ticks.saturating_mul(STATE_TRANSFER_OFFER_EXPIRY_MULTIPLE);
let served_expiry_ticks = retry_ticks.saturating_mul(STATE_TRANSFER_SERVED_EXPIRY_MULTIPLE);
+ // A build slot is released by the round that completes the offer, so a
+ // requester that walked away mid-build would otherwise hold admission
+ // forever. Same idle window as an abandoned offer.
+ self.partition_offer_builds
+ .borrow_mut()
+ .retain(|namespace, idle_ticks| {
+ *idle_ticks += 1;
+ let live = *idle_ticks < idle_expiry_ticks;
+ if !live {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace,
+ "dropping an abandoned partition offer build slot"
+ );
+ }
+ live
+ });
let mut offers = self.state_transfer_offers.borrow_mut();
let namespaces_before: Vec<u64> = offers.keys().map(|(namespace, _)| *namespace).collect();
offers.retain(|(namespace, requester), served| {
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index ce153e957..955353200 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -698,15 +698,41 @@ where
"purge-partition reset partition to empty"
);
}
- Err(error) => {
- // Fence this one group for rebuild, exactly as a
- // failed state-transfer convergence does. Both
- // failure points need it for different reasons: the
- // frontier reset fails before anything is mutated
- // and leaves a partition still holding data the
- // group believes it purged, while everything after
- // the drain leaves no serviceable chain at all and
- // the next append panics on `active_segment()`.
+ Err(partitions::PurgeError::FrontierNotRecorded) => {
+ // NOT fenced: nothing was mutated, so the chain is
+ // whole and `applied_purge_generation` is unmoved,
+ // which means the reconciler's `committed > applied`
+ // gate still sees this purge as outstanding.
+ // Fencing here would quarantine live data, and the
+ // fence's own frontier write would first stamp the
+ // pre-purge counter the purge was about to reset.
+ //
+ // NOT woken: staging a purge counts as work in the
+ // pass, which keeps the fast-skip disarmed, so the
+ // ordinary periodic pass re-issues until one lands
+ // and stops once `applied` catches `committed`. An
+ // eager wake here closes a loop with no pacing in
+ // it at all -- pass, stage, defer, wake -- and on a
+ // disk that refuses instantly that is a full O(N)
+ // reconcile scan and a real `atomic_replace`
+ // attempt per turn, holding the partition write
+ // lock each time.
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ generation,
+ "purge-partition deferred: could not record the frontier reset; \
+ the reconciler re-issues it while the generation stays unapplied"
+ );
+ }
+ Err(error @ partitions::PurgeError::Unserviceable(_)) => {
+ // Past the drain, so this group has no serviceable
+ // chain and the next append panics on
+ // `active_segment()`. Fence it for rebuild, exactly
+ // as a failed state-transfer convergence does. The
+ // counters were already reset to 0 before the
+ // fallible plant, so the fence's advancing write
+ // records the post-purge frontier.
tracing::error!(
shard = self.id,
namespace_raw = namespace.inner(),
@@ -717,7 +743,8 @@ where
// Fenced, but the caches still describe the
// pre-purge bytes until the rebuild lands.
self.drop_partition_transfer_state(namespace, partition);
- self.fence_partition_for_rebuild(namespace, partition).await;
+ self.fence_partition_for_rebuild(namespace, partition, None)
+ .await;
}
}
}