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 <[email protected]> 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(),
