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:52 UTC
(iggy) branch partition_superblock updated (ed61c3e51 -> b462ce356)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a change to branch partition_superblock
in repository https://gitbox.apache.org/repos/asf/iggy.git
from ed61c3e51 address review comments
new 85c755b5d fix some criticial bugs
new b462ce356 review changes
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of 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 | 542 +++++++++++++++++++++++--
core/partitions/src/journal.rs | 7 +-
core/partitions/src/lib.rs | 2 +-
core/partitions/src/state_transfer.rs | 270 ++++++++++--
core/server-ng/config.toml | 22 +-
core/server-ng/src/bootstrap.rs | 7 +-
core/server-ng/src/partition_helpers.rs | 44 +-
core/server-ng/src/partition_reconciler.rs | 26 +-
core/server-ng/src/server_error.rs | 2 +-
core/shard/src/lib.rs | 469 +++++++++++++++++----
core/shard/src/router.rs | 45 +-
15 files changed, 1264 insertions(+), 219 deletions(-)
(iggy) 02/02: review changes
Posted by gk...@apache.org.
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;
}
}
}
(iggy) 01/02: fix some criticial bugs
Posted by gk...@apache.org.
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 85c755b5d12a9e41f3d6b3b31a76613ae279a9f6
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Thu Aug 6 12:38:33 2026 +0200
fix some criticial bugs
---
core/partitions/src/iggy_partition.rs | 210 +++++++++++++++++++++++++++++---
core/partitions/src/state_transfer.rs | 120 +++++++++++++++---
core/server-ng/src/partition_helpers.rs | 7 ++
core/shard/src/lib.rs | 76 +++++++++++-
core/shard/src/router.rs | 14 ++-
5 files changed, 385 insertions(+), 42 deletions(-)
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 0981bf093..bc6d1c246 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -79,7 +79,7 @@ use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::Mutex as TokioMutex;
-use tracing::{debug, warn};
+use tracing::{debug, error, warn};
// This struct aliases in terms of the code contained the `LocalPartition from `core/server/src/streaming/partitions/local_partition.rs`.
//
@@ -180,6 +180,17 @@ where
/// terminal policy.
superblock_write_failures: Cell<u64>,
superblock_retry_after_micros: Cell<u64>,
+ /// The `offset_frontier` the last successful superblock write recorded,
+ /// seeded at boot from the record that write left behind.
+ ///
+ /// The advance direction maxes against THIS as well as the live counter,
+ /// because the two diverge: a failed install leaves the counter at its
+ /// pre-install value while the record already names the incoming frontier,
+ /// and the fence that follows then persists the counter. Maxing against
+ /// the counter alone writes 0 over a recorded N and quarantines the
+ /// segments that were the only other witness, after which the rebuild
+ /// re-mints offsets the group already handed out.
+ durable_offset_frontier: Cell<u64>,
/// In-flight state transfer for this group (rejoin whose repair floor was
/// refused); tail repair takes over at install. See
/// [`PartitionTransferSession`].
@@ -365,6 +376,7 @@ where
superblock_lock: LocalGate::new(),
superblock_write_failures: Cell::new(0),
superblock_retry_after_micros: Cell::new(0),
+ durable_offset_frontier: Cell::new(0),
transfer: None,
transfer_attempts: 0,
transfer_failures: 0,
@@ -501,9 +513,15 @@ where
/// 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, offset_frontier: u64) -> bool {
- // ADVANCE direction: never below what this replica has already minted.
- // The reset direction (purge) goes through `write_superblock_inner`.
- let advanced = offset_frontier.max(self.offset_frontier());
+ // ADVANCE direction: never below what this replica has already minted,
+ // and never below what the record ALREADY holds. Both bounds are
+ // needed and neither implies the other -- a failed install leaves the
+ // counter behind the record it wrote before the swap, so maxing against
+ // the counter alone lets the fence that follows lower the durable
+ // frontier. The reset direction goes through `write_superblock_inner`.
+ let advanced = offset_frontier
+ .max(self.offset_frontier())
+ .max(self.durable_offset_frontier.get());
self.write_superblock_inner(superblock, advanced).await
}
@@ -529,6 +547,7 @@ where
Ok(()) => {
self.consensus
.mark_superblock_durable(state.view, state.log_view);
+ self.durable_offset_frontier.set(state.offset_frontier);
self.superblock_write_failures.set(0);
self.superblock_retry_after_micros.set(0);
true
@@ -603,15 +622,37 @@ where
/// peer stamps 0.
#[allow(clippy::future_not_send)]
pub async fn reset_offset_frontier(&self) -> bool {
+ self.reset_offset_frontier_to(self.offset_frontier()).await
+ }
+
+ /// [`Self::reset_offset_frontier`] for a frontier the live counter does not
+ /// hold yet.
+ ///
+ /// Two callers need the value spelled out rather than read off the counter.
+ /// A purge records its reset BEFORE it unlinks anything, while the counter
+ /// still names the pre-purge space, so a crash mid-unlink cannot boot into
+ /// a re-seed of the space the purge was erasing. An install under an
+ /// advancing purge generation records the offer's frontier, which is
+ /// legitimately below the local counter: the advancing form would max it
+ /// 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 {
let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
return true;
};
let _superblock_guard = self.superblock_lock.acquire().await;
- let frontier = self.offset_frontier();
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);
+ }
+
/// [`Self::persist_offset_frontier`] for a frontier this replica has not
/// reached yet.
///
@@ -3450,6 +3491,41 @@ where
Ok(())
}
+ /// Record the purge's frontier reset BEFORE the purge touches anything.
+ ///
+ /// The unlinks are made durable by their own directory fsync, so a crash
+ /// between them and a reset written afterwards boots a purged directory
+ /// whose record still names the pre-purge offset space:
+ /// `restore_offset_frontier` re-seeds the counter to it while every peer
+ /// restarted at 0, and the first append stamps a `base_offset` and
+ /// `batch_checksum` no peer shares. Writing 0 first inverts the window into
+ /// a harmless one -- the record under-claims while the segments still
+ /// exist, and boot takes the max of the record and what the segments prove.
+ ///
+ /// Spelled out rather than read off the counter, which still holds the
+ /// 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.
+ #[allow(clippy::future_not_send)]
+ async fn record_purge_frontier_reset(&self, generation: u64) -> Result<(), IggyError> {
+ if self.reset_offset_frontier_to(0).await {
+ return Ok(());
+ }
+ error!(
+ 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"
+ );
+ Err(IggyError::Error)
+ }
+
/// Reset the partition to a single empty segment at offset 0 and clear all
/// consumer / consumer-group offsets (memory + disk). This is the local
/// effect of a committed `PurgeTopic`: it wipes message data and offsets but
@@ -3461,12 +3537,14 @@ where
/// `PurgeTopic` advances the committed generation and triggers a fresh pass).
///
/// # Errors
- /// If the replacement segment's log / index file cannot be created. Every
- /// segment is already drained by then, so an `Err` leaves a partition with
- /// no serviceable chain: the caller must FENCE it (quarantine + retire for
- /// the reconciler to rebuild), exactly as the state-transfer install's
- /// `ConvergeFailed` arm does, or the next append panics on
- /// `active_segment()`.
+ /// 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
+ /// 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.
+ #[allow(clippy::too_many_lines)]
pub async fn purge(
&mut self,
config: &PartitionsConfig,
@@ -3477,6 +3555,8 @@ where
let namespace = self.namespace();
+ self.record_purge_frontier_reset(generation).await?;
+
// The purge recreates segment files at the paths it unlinks below, so
// an in-flight poll's cached read fd would keep serving the unlinked
// pre-purge inodes as live data. Wipe the shared read-state slots
@@ -3626,12 +3706,20 @@ where
// Same commit frontier, different (now empty) bytes: a cached offer
// built pre-purge would advertise files the purge just unlinked.
self.transfer_offer_cache.borrow_mut().take();
- // RESET, not advance: the durable frontier still names the pre-purge
- // offset space, and leaving it there makes the next boot re-seed the
- // counter to the state this purge just erased -- after which the first
- // append stamps `base_offset` N while every peer stamps 0. The live
- // counter is 0 by now, so the reset records 0.
- self.reset_offset_frontier().await;
+ // The reset itself already landed before the unlinks; this second write
+ // only re-stamps the record now that the view-scoped fields and the
+ // counter agree with it. A failure leaves the pre-unlink 0 on disk,
+ // which is the safe direction, so it is logged rather than refused.
+ if !self.reset_offset_frontier().await {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ generation,
+ "purge could not re-stamp the superblock after resetting the partition; \
+ the frontier reset written before the unlinks still stands"
+ );
+ }
Ok(())
}
@@ -4256,6 +4344,94 @@ mod tests {
);
}
+ /// The `offset_frontier` of the most recent recorded write.
+ fn last_recorded_frontier(store: &RecordingSuperblock) -> u64 {
+ let writes = store.writes.borrow();
+ let bytes = writes.last().expect("a superblock write landed");
+ consensus::VsrState::try_from(bytes.as_slice())
+ .expect("recorded payload decodes as a VsrState")
+ .offset_frontier
+ }
+
+ /// The fence path persists the frontier while the live counter still sits
+ /// at its pre-install value, so an advance that maxes against the counter
+ /// alone erases the record and then quarantines the segments that were its
+ /// only other witness. Boot re-mints from 0 against a group at N after that.
+ #[compio::test]
+ 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());
+
+ assert!(partition.persist_offset_frontier_at(9_000).await);
+ assert_eq!(last_recorded_frontier(&store), 9_000);
+ assert_eq!(
+ partition.offset_frontier(),
+ 0,
+ "a partition that never minted reports a zero frontier, which is the \
+ value the fence would otherwise persist"
+ );
+
+ assert!(partition.persist_offset_frontier().await);
+
+ assert_eq!(
+ last_recorded_frontier(&store),
+ 9_000,
+ "the advance direction must not lower the durable frontier"
+ );
+ }
+
+ /// 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.
+ #[compio::test]
+ 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.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_eq!(
+ last_recorded_frontier(&store),
+ 12,
+ "the reset must record the incoming frontier, not max back up to the \
+ counter the install is about to replace"
+ );
+ }
+
+ /// A purge records its reset before it unlinks anything, so a write it
+ /// cannot make has to stop the purge while the data proving the old
+ /// frontier is still on disk.
+ #[compio::test]
+ 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.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());
+
+ store.fail_writes.set(false);
+ partition
+ .record_purge_frontier_reset(7)
+ .await
+ .expect("a working store records the reset");
+ assert_eq!(
+ last_recorded_frontier(&store),
+ 0,
+ "the reset is spelled out, not read off a counter still holding the \
+ pre-purge frontier"
+ );
+ }
+
#[compio::test]
async fn given_undurable_view_when_sending_prepare_ok_should_withhold_until_persisted() {
let bus = RecordingBus::default();
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 8f943f0e2..7ae3c8ca8 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -926,6 +926,13 @@ pub enum PartitionTransferUnavailable {
/// The segment chain changed while the offer's checksum passes ran, so the
/// stamps no longer describe the bytes the offer addresses.
SegmentSetChanged,
+ /// This round's share of the checksum pass ran out with retained bytes
+ /// still unhashed. Progress is memoized, so the next request resumes where
+ /// this one stopped rather than starting the pass again.
+ OfferBuildInProgress {
+ hashed: u64,
+ remaining: u64,
+ },
FlushFailed(iggy_common::IggyError),
SegmentUnreadable {
start_offset: u64,
@@ -946,7 +953,8 @@ impl PartitionTransferUnavailable {
Self::NotCaughtUpPrimary
| Self::RepairInProgress
| Self::NothingCommitted
- | Self::SegmentSetChanged => true,
+ | Self::SegmentSetChanged
+ | Self::OfferBuildInProgress { .. } => true,
Self::NoPartitionDir
| Self::ManifestTooLarge { .. }
| Self::FlushFailed(_)
@@ -972,6 +980,11 @@ impl fmt::Display for PartitionTransferUnavailable {
Self::SegmentSetChanged => {
write!(f, "segment chain changed while the offer was being built")
}
+ Self::OfferBuildInProgress { hashed, remaining } => write!(
+ f,
+ "offer checksum pass is {hashed} bytes in with {remaining} to go; \
+ resuming on the next request"
+ ),
Self::FlushFailed(source) => {
write!(f, "flushing the committed prefix failed: {source}")
}
@@ -1398,11 +1411,34 @@ where
});
}
+ // The checksum pass is the expensive part and it runs inside ONE frame
+ // body: the router's tick arm is not polled while another arm's body
+ // awaits, and the yields inside `hash_segment_range` move the reactor,
+ // not this shard's consensus ticks. A cold pass over multi-GiB
+ // retention therefore silences every group on this core for its whole
+ // duration, past `heartbeat_timeout`, on the node that by construction
+ // is the caught-up primary of those groups.
+ //
+ // Bounded per round instead. The memo carries partial progress, so a
+ // refusal here is not lost work: the requester re-asks on its flat
+ // transient interval and each round advances the pass by the budget
+ // until the offer completes.
+ let mut budget = OFFER_HASH_BUDGET_PER_ROUND_BYTES;
let mut segments = Vec::with_capacity(planned.len());
- for (start_offset, size, log_path) in &planned {
- let checksum = self
- .segment_checksum(*start_offset, *size, log_path)
- .await?;
+ for (index, (start_offset, size, log_path)) in planned.iter().enumerate() {
+ 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>();
+ return Err(PartitionTransferUnavailable::OfferBuildInProgress {
+ hashed: OFFER_HASH_BUDGET_PER_ROUND_BYTES.saturating_sub(budget),
+ remaining,
+ });
+ };
segments.push(SegmentArtifactSource {
entry: consensus::StateArtifact {
kind: artifact_kind::SEGMENT_LOG,
@@ -1469,6 +1505,10 @@ where
/// retained history every round: `commit_op` advances per round, so the
/// offer cache misses even when nothing else changed.
///
+ /// `budget` caps the bytes this call may read, charged as it goes. `None`
+ /// means the budget ran out first: the memo holds everything hashed so far
+ /// and the next call resumes from it.
+ ///
/// # Errors
/// [`PartitionTransferUnavailable::SegmentUnreadable`] when the file is
/// unreadable or shorter than `size`.
@@ -1477,7 +1517,8 @@ where
start_offset: u64,
size: u64,
log_path: &str,
- ) -> Result<u64, PartitionTransferUnavailable> {
+ budget: &mut u64,
+ ) -> Result<Option<u64>, PartitionTransferUnavailable> {
// Taken OUT of the map for the read: the hash awaits, and a half-fed
// hasher left visible could be extended twice by a second build.
let memo = self
@@ -1503,17 +1544,28 @@ where
SegmentChecksumMemo::new()
}
};
- hash_segment_range(log_path, memo.hashed_len, size, &mut memo.hasher, None)
+ // Clamped to the round's remaining budget, so a single multi-GiB
+ // segment is split across rounds rather than being the granularity
+ // floor. `finish` does not consume the hasher, so a partial pass is
+ // simply a memo nobody stamps yet.
+ let target = size.min(memo.hashed_len.saturating_add(*budget));
+ let hashed = target.saturating_sub(memo.hashed_len);
+ // Dropped on failure, not reinserted: the hasher is fed chunk by chunk
+ // and a mid-range error leaves it holding bytes `hashed_len` does not
+ // account for, so resuming from it would stamp a checksum over a
+ // doubly-fed prefix. Losing the partial pass is the cheap side.
+ hash_segment_range(log_path, memo.hashed_len, target, &mut memo.hasher, None)
.await
.map_err(|source| PartitionTransferUnavailable::SegmentUnreadable {
start_offset,
source,
})?;
- memo.hashed_len = size;
- let checksum = memo.hasher.finish();
+ memo.hashed_len = target;
+ let checksum = (target == size).then(|| memo.hasher.finish());
self.segment_checksum_cache
.borrow_mut()
.insert(start_offset, memo);
+ *budget = budget.saturating_sub(hashed);
Ok(checksum)
}
@@ -1872,12 +1924,22 @@ where
// wedge the in-memory case. Nothing has been mutated yet.
//
// Under `purge_advances` the offer's frontier is legitimately BELOW the
- // live counter; the advancing write maxes it back up, which is correct
- // here -- the reset belongs to `purge`, which records 0 as it runs.
- if !self
- .persist_offset_frontier_at(offsets_wire.next_offset)
- .await
- {
+ // live counter and must be written as a RESET. The advancing form would
+ // max it back up to the pre-purge value, and the reset it defers to
+ // belongs to a `purge` this replica provably never ran -- `purge_advances`
+ // is true precisely because it missed one. A crash between the old
+ // chain's unlink fsync and the last staged rename would then boot with
+ // zero `.log` files and re-seed the counter from the pre-purge frontier,
+ // 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)
+ .await
+ } else {
+ self.persist_offset_frontier_at(offsets_wire.next_offset)
+ .await
+ };
+ if !frontier_durable {
return Err(PartitionInstallError::FrontierNotDurable {
frontier: offsets_wire.next_offset,
});
@@ -1918,7 +1980,22 @@ where
// 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;
+ //
+ // Logged rather than refused: the install already mutated, and the
+ // pre-swap write above left a valid lower bound on disk either way. The
+ // failure still matters -- the ordinary retry is the view-change gate,
+ // which an idle group may not reach for a long time -- so it must not
+ // pass silently.
+ if !self.persist_offset_frontier().await {
+ tracing::error!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.consensus().namespace(),
+ frontier = self.offset_frontier(),
+ "state-transfer install could not record the installed offset frontier; \
+ the durable record stays at the pre-swap claim until the next view change"
+ );
+ }
outcome
}
@@ -2569,6 +2646,17 @@ async fn yield_to_reactor() {
/// the reactor many times per segment.
const OFFER_HASH_CHUNK_LEN: usize = 1 << 20;
+/// Bytes one offer-build round may read and hash before it refuses and resumes
+/// on the next request.
+///
+/// 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.
+const OFFER_HASH_BUDGET_PER_ROUND_BYTES: u64 = 256 * 1024 * 1024;
+
/// Feed bytes `[from, to)` of `path` into `hasher`, read in
/// [`OFFER_HASH_CHUNK_LEN`] chunks with one reactor yield per chunk, appending
/// each chunk to `sink` when one is given.
diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs
index ac0200460..2f217004a 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -567,6 +567,13 @@ 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)
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index dd54e8d7c..c4dff2349 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -893,9 +893,16 @@ pub const SERVED_SEGMENT_CACHE_BYTES_DEFAULT: u64 =
/// TWO, not the receiver's in-flight cap of four: the budget is PER SHARD and
/// shard count defaults to core count, so each segment here multiplies by the
/// core count during a whole-node rejoin, on top of page cache and the receive
-/// side's own in-flight artifacts. Two keeps one pull's segment resident while a
-/// second rotates through; running under the budget costs re-reads, not
-/// failures, and operators serving many concurrent rejoins raise the knob.
+/// side's own in-flight artifacts.
+///
+/// The gap between this and the in-flight cap is closed by ADMITTING fewer
+/// concurrent transfers rather than by holding more bytes: see
+/// `IggyShard::partition_transfer_admission_cap`, which derives its cap from
+/// this budget so the two can never disagree. Overrunning the budget does not
+/// degrade gracefully -- distinct groups are distinct cache keys, so a surplus
+/// pull evicts the others on every chunk and none of them converge -- and an
+/// operator who wants more concurrency raises the knob, which raises the cap
+/// with it.
const CONCURRENT_SERVED_SEGMENTS: u64 = 2;
/// Shard-wide cache of segment payloads loaded to serve partition chunks,
@@ -5587,6 +5594,39 @@ where
}
}
+ /// Whether this shard may build a partition offer for `namespace` without
+ /// pushing the served-payload working set past its byte budget.
+ ///
+ /// Counts DISTINCT groups rather than requesters: the payload cache is
+ /// content-addressed, so every requester pulling one group's offer shares
+ /// one resident copy, and it is the group count that decides how many
+ /// 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.
+ fn partition_transfer_admission_cap(&self) -> usize {
+ let slots = self.served_segment_cache_bytes_max.get() / SEGMENT_SIZE_CEILING_BYTES;
+ usize::try_from(slots).unwrap_or(usize::MAX).max(1)
+ }
+
+ fn may_serve_another_partition_transfer(&self, namespace: u64) -> bool {
+ let offers = self.state_transfer_offers.borrow();
+ let mut served: Vec<u64> = offers
+ .iter()
+ .filter(|(_, served)| matches!(served.offer, ServedOffer::Partition(_)))
+ .map(|((offer_namespace, _), _)| *offer_namespace)
+ .collect();
+ if served.contains(&namespace) {
+ return true;
+ }
+ served.sort_unstable();
+ served.dedup();
+ served.len() < self.partition_transfer_admission_cap()
+ }
+
/// 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)]
@@ -5644,6 +5684,36 @@ where
);
Some(offer)
}
+ None if !self.may_serve_another_partition_transfer(header.namespace) => {
+ // Admission control, because the served-payload budget is a
+ // BYTE budget and the pulls that overrun it do not degrade
+ // gracefully. Each concurrent pull holds a different segment
+ // resident, so admitting more distinct groups than the budget
+ // has max-size slots makes them evict each other on every
+ // chunk: every request then re-reads and re-hashes a whole
+ // segment to serve one 256 KiB range, and a per-chunk serve
+ // that outruns the requester's stall interval exhausts its
+ // retry budget, so the pull rotates peers and never converges.
+ // Refusing the surplus is what makes the admitted ones finish.
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ requester = header.replica,
+ "already serving as many partition transfers as the served-payload \
+ budget holds; refusing until one completes"
+ );
+ let (view, commit_max) = serving_progress(partition);
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ TransferDescriptor::unavailable(true, view, commit_max),
+ )
+ .await;
+ return;
+ }
None => match partition.state_transfer_offer(&config).await {
Ok(offer) => {
tracing::info!(
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index c426e526a..ce153e957 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -699,12 +699,14 @@ where
);
}
Err(error) => {
- // The purge drained every segment before its first
- // fallible step, so a failure leaves the partition
- // with no serviceable chain and the next append
- // panics on `active_segment()`. Fence this one group
- // for rebuild, exactly as a failed state-transfer
- // convergence does.
+ // 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()`.
tracing::error!(
shard = self.id,
namespace_raw = namespace.inner(),