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 <[email protected]> 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; } } }
