This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch offset_hardening in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 5bc3ced7e3308af3146355f06ad29b6bb4661107 Author: spetz <[email protected]> AuthorDate: Mon Sep 7 10:39:07 2026 +0200 hardening --- core/common/src/traits/message_client.rs | 5 + core/configs/src/server_config/partition.rs | 4 +- .../tests/server/consumer_offset_quota_vsr.rs | 48 +- core/partitions/src/consumer_offset_capacity.rs | 56 ++- core/partitions/src/iggy_partition.rs | 528 +++++++++++++++++---- core/partitions/src/offset_storage.rs | 17 +- core/partitions/src/poll_plan.rs | 32 +- core/partitions/src/state_transfer.rs | 37 +- core/sdk/src/clients/consumer.rs | 15 +- core/sdk/src/leader_aware.rs | 9 +- core/sdk/src/tcp/tcp_client.rs | 16 +- core/server/config.toml | 15 +- core/server/src/boot/recovery.rs | 4 +- core/server/src/dispatch/partition.rs | 109 +++-- core/server/src/offset_recovery.rs | 118 ++++- core/server/src/partition_reconciler.rs | 14 + core/shard/src/lib.rs | 35 +- core/shard/src/metrics.rs | 49 +- core/simulator/src/lib.rs | 2 +- 19 files changed, 870 insertions(+), 243 deletions(-) diff --git a/core/common/src/traits/message_client.rs b/core/common/src/traits/message_client.rs index de35e454d..b71017647 100644 --- a/core/common/src/traits/message_client.rs +++ b/core/common/src/traits/message_client.rs @@ -35,6 +35,11 @@ pub trait MessageClient { /// rejected with `TooManyConsumerOffsets` at the partition's configured /// limit. That poll returns no messages. Existing keys remain writable, /// and polling without auto-commit does not allocate a stored offset. + /// A refused auto-commit submission returns `TransientNotAccepted` with no + /// messages and may be retried. A capacity error requires capacity to be freed. + /// Local cursors without a committed offset can be evicted at the limit. + /// Their next `Next` poll resumes from the earliest retained messages, + /// which can redeliver messages from earlier polls. #[allow(clippy::too_many_arguments)] async fn poll_messages( &self, diff --git a/core/configs/src/server_config/partition.rs b/core/configs/src/server_config/partition.rs index eb2275ca6..56257936c 100644 --- a/core/configs/src/server_config/partition.rs +++ b/core/configs/src/server_config/partition.rs @@ -186,8 +186,8 @@ pub struct PartitionConfig { /// Whether consumer-offset files are written crash-safe: data-synced, then /// renamed over the prior cursor, with the directory synced once per commit /// walk. Independent of the topic's `enforce_fsync`, which governs message - /// and index files. Off, an offset file is rewritten in place with no sync - /// and a crash costs at most a redelivery from the last flushed cursor. + /// and index files. Off, an offset file is rewritten in place with no sync. + /// A lost or torn cursor can cause replay from the earliest retained data. #[serde(default)] pub consumer_offset_enforce_fsync: bool, diff --git a/core/integration/tests/server/consumer_offset_quota_vsr.rs b/core/integration/tests/server/consumer_offset_quota_vsr.rs index 7cc301c14..ef454b81e 100644 --- a/core/integration/tests/server/consumer_offset_quota_vsr.rs +++ b/core/integration/tests/server/consumer_offset_quota_vsr.rs @@ -347,18 +347,42 @@ async fn given_full_consumer_offset_table_when_creating_another_should_reject_wi .text() .await .expect("metrics text"); - let denied: u64 = metrics - .lines() - .filter(|line| line.starts_with("partition_consumer_offsets_denied_total{")) - .map(|line| { - line.split_whitespace() - .last() - .expect("counter value") - .parse::<u64>() - .expect("numeric counter") - }) - .sum(); - assert_eq!(denied, 3, "one explicit TCP, one poll, and one HTTP denial"); + let denied_for = |kind: &str| -> u64 { + let kind_label = format!("kind=\"{kind}\""); + let mut samples = 0; + let total = metrics + .lines() + .filter_map(|line| { + line.strip_prefix("partition_consumer_offsets_denied_total{")? + .split_once('}') + }) + .filter(|(labels, _)| labels.split(',').any(|label| label == kind_label)) + .map(|(_, value)| { + samples += 1; + value + .split_whitespace() + .last() + .expect("counter value") + .parse::<u64>() + .expect("numeric counter") + }) + .sum(); + assert!( + samples > 0, + "missing {kind} denial metric in response: {metrics}" + ); + total + }; + assert_eq!( + denied_for("consumer"), + 3, + "one explicit TCP, one poll, and one HTTP denial, all on the consumer kind" + ); + assert_eq!( + denied_for("consumer_group"), + 0, + "no consumer group offset was denied in this test" + ); } #[iggy_harness( diff --git a/core/partitions/src/consumer_offset_capacity.rs b/core/partitions/src/consumer_offset_capacity.rs index e2ae15c9a..aaaa93208 100644 --- a/core/partitions/src/consumer_offset_capacity.rs +++ b/core/partitions/src/consumer_offset_capacity.rs @@ -173,6 +173,7 @@ pub struct ConsumerOffsetCapacity { limit: Cell<usize>, pending: RefCell<HashMap<u32, usize>>, provisional: RefCell<HashMap<u32, Arc<ProvisionalToken>>>, + active_provisional_keys: Arc<AtomicUsize>, stranded: RefCell<HashSet<u32>>, uncertain: Cell<bool>, durable_warned: Cell<bool>, @@ -188,6 +189,7 @@ impl ConsumerOffsetCapacity { limit: Cell::new(limit), pending: RefCell::new(HashMap::new()), provisional: RefCell::new(HashMap::new()), + active_provisional_keys: Arc::new(AtomicUsize::new(0)), stranded: RefCell::new(HashSet::new()), uncertain: Cell::new(false), durable_warned: Cell::new(false), @@ -225,10 +227,11 @@ impl ConsumerOffsetCapacity { } let limit = self.limit.get(); let durable_count = durable.count(self.kind); - let upper_bound = durable_count + let fixed = durable_count .saturating_add(self.pending.borrow().len()) - .saturating_add(self.provisional.borrow().len()) .saturating_add(self.stranded.borrow().len()); + let provisional_len = self.active_provisional_keys.load(Ordering::Relaxed); + let upper_bound = fixed.saturating_add(provisional_len); if !self.uncertain.get() && upper_bound < limit { self.durable_warned.set(false); return Ok(()); @@ -237,12 +240,6 @@ impl ConsumerOffsetCapacity { let occupied = if durable_count >= limit { durable_count } else { - let mut provisional = self.provisional.borrow_mut(); - if provisional.len() >= limit { - provisional - .retain(|key, token| *key == id || token.active.load(Ordering::Relaxed) > 0); - } - drop(provisional); self.occupied(durable) }; if self.uncertain.get() || occupied >= limit { @@ -265,13 +262,19 @@ impl ConsumerOffsetCapacity { ) -> Result<AutoCommitReservation, ConsumerOffsetCapacityError> { self.check(id, durable)?; let mut provisional = self.provisional.borrow_mut(); + if provisional.len() >= self.limit.get() && !provisional.contains_key(&id) { + provisional.retain(|_, token| token.active.load(Ordering::Relaxed) > 0); + } let token = Arc::clone(provisional.entry(id).or_insert_with(|| { Arc::new(ProvisionalToken { reclaim_epoch: Arc::clone(&self.reclaim_epoch), + active_keys: Arc::clone(&self.active_provisional_keys), active: AtomicUsize::new(0), }) })); - token.active.fetch_add(1, Ordering::Relaxed); + if token.active.fetch_add(1, Ordering::Relaxed) == 0 { + token.active_keys.fetch_add(1, Ordering::Relaxed); + } Ok(AutoCommitReservation { token, kind: self.kind, @@ -298,6 +301,10 @@ impl ConsumerOffsetCapacity { .is_some_and(|token| token.active.load(Ordering::Relaxed) > 0) } + /// Assigns the pending count outright while [`Self::release_reservation`] + /// decrements it. Both take `&self` and neither locks: they are serialized + /// by their call sites, which all run under the partition's `&mut self` on + /// its own shard thread. pub(crate) fn set_pending_count(&self, id: u32, count: usize) { if count == 0 { if self.pending.borrow_mut().remove(&id).is_some() { @@ -308,6 +315,7 @@ impl ConsumerOffsetCapacity { } } + /// See [`Self::set_pending_count`] for the serialization contract. pub(crate) fn release_reservation(&self, id: u32) { let mut pending = self.pending.borrow_mut(); let Some(count) = pending.get_mut(&id) else { @@ -359,6 +367,14 @@ impl ConsumerOffsetCapacity { self.stranded.borrow().contains(&id) } + /// Keys whose file could not be loaded or unlinked. Cleared only by a + /// later store or delete of the same key, never by `rebuild` or + /// `mark_uncertain`, so a permanently unwritable file keeps this above + /// zero. Exported as a gauge so that refusal has a signal. + pub(crate) fn stranded_count(&self) -> usize { + self.stranded.borrow().len() + } + pub(crate) fn rearm_if_below_limit(&self, durable: &DurableConsumerOffsets) { if !self.durable_warned.get() || self.uncertain.get() @@ -448,12 +464,14 @@ pub struct AutoCommitReservation { #[derive(Debug)] struct ProvisionalToken { reclaim_epoch: Arc<AtomicU64>, + active_keys: Arc<AtomicUsize>, active: AtomicUsize, } impl Drop for AutoCommitReservation { fn drop(&mut self) { if self.token.active.fetch_sub(1, Ordering::Relaxed) == 1 { + self.token.active_keys.fetch_sub(1, Ordering::Relaxed); self.token.reclaim_epoch.fetch_add(1, Ordering::Relaxed); } } @@ -463,6 +481,26 @@ impl Drop for AutoCommitReservation { mod tests { use super::*; + #[test] + fn given_cached_inactive_tokens_when_admitting_should_count_only_active_keys() { + let durable = Rc::new(DurableConsumerOffsets::default()); + let capacity = Rc::new(ConsumerOffsetCapacity::new(ConsumerKind::Consumer, 2)); + let first = capacity.reserve_provisional(1, &durable).unwrap(); + let repeated = capacity.reserve_provisional(1, &durable).unwrap(); + assert_eq!(capacity.active_provisional_keys.load(Ordering::Relaxed), 1); + drop(first); + assert_eq!(capacity.active_provisional_keys.load(Ordering::Relaxed), 1); + drop(repeated); + assert_eq!(capacity.active_provisional_keys.load(Ordering::Relaxed), 0); + assert_eq!(capacity.provisional.borrow().len(), 1); + capacity.check(2, &durable).unwrap(); + let second = capacity.reserve_provisional(2, &durable).unwrap(); + assert_eq!(capacity.active_provisional_keys.load(Ordering::Relaxed), 1); + drop(second); + capacity.check(3, &durable).unwrap(); + assert_eq!(capacity.active_provisional_keys.load(Ordering::Relaxed), 0); + } + #[test] fn given_low_occupancy_when_accounting_is_uncertain_should_reject_new_keys() { let durable = DurableConsumerOffsets::default(); diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 5e4d964d8..1fa2e9646 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -222,9 +222,11 @@ where offset_reservations_scan_state: Option<(u64, u64, u64, Option<u64>)>, consumer_group_offsets_reconcile_epoch: Rc<Cell<u64>>, consumer_offset_dirs_dirty: [Cell<bool>; 2], - /// Set once the offsets mount refused a directory fsync, so the warning - /// fires once per partition rather than per commit walk. - consumer_offset_dir_sync_unsupported: Cell<bool>, + /// Latest operation of each kind in the current commit walk. Even a covered + /// store depends on an earlier unsynced directory entry of its kind. + consumer_offset_dirs_touched: [Cell<Option<(u64, Operation)>>; 2], + #[cfg(test)] + consumer_offset_dir_sync_fault: Cell<Option<usize>>, #[cfg(test)] offset_dir_sync_count: Cell<usize>, /// Highest `PurgeTopic` generation this replica has locally applied (reset @@ -579,7 +581,9 @@ where offset_reservations_scan_state: None, consumer_group_offsets_reconcile_epoch: Rc::new(Cell::new(0)), consumer_offset_dirs_dirty: [Cell::new(false), Cell::new(false)], - consumer_offset_dir_sync_unsupported: Cell::new(false), + consumer_offset_dirs_touched: [Cell::new(None), Cell::new(None)], + #[cfg(test)] + consumer_offset_dir_sync_fault: Cell::new(None), #[cfg(test)] offset_dir_sync_count: Cell::new(0), applied_purge_generation: 0, @@ -2122,6 +2126,13 @@ where // durably stored; the in-memory update is idempotent on replay // because we look up by (kind, id). self.persist_consumer_offset_commit(pending).await?; + let operation = match pending.mutation { + PendingConsumerOffsetMutation::Upsert(_) => Operation::StoreConsumerOffset, + PendingConsumerOffsetMutation::Delete => Operation::DeleteConsumerOffset, + }; + // A covered store also depends on an earlier dirty directory entry. + self.consumer_offset_dirs_touched[crate::state_transfer::consumer_kind_index(pending.kind)] + .set(Some((op, operation))); self.apply_consumer_offset_commit(pending); self.pending_consumer_offset_commits.remove(&op); self.refresh_consumer_offset_reservation(pending.kind, pending.consumer_id); @@ -2206,10 +2217,9 @@ where } PendingConsumerOffsetMutation::Delete => { if let Some(path) = path.as_deref() { - // A committed delete applies on every replica, so an - // unlinkable file cannot refuse it. The key stays stranded: - // the file may resurrect at boot, and a stranded key admits - // the delete that settles it. + // Keep the logical state until the file is removed. The + // partition journal is memory-only, so acknowledging an + // unsuccessful unlink would let boot resurrect the key. match delete_persisted_offset(path).await { Ok(removed) => { if removed && self.consumer_offset_enforce_fsync { @@ -2230,6 +2240,7 @@ where %error, "committed consumer offset delete could not remove its file" ); + return Err(error); } } } @@ -2355,21 +2366,27 @@ where if !self.consensus.is_primary() || !self.consensus.is_normal() { return Vec::new(); } + // A stranded file already failed normal loading or unlink. Reissuing + // replicated deletes every reconciliation pass cannot make its + // filesystem writable and would create a permanent commit loop. + // A single-replica explicit deletion can retry after repair. On a + // replicated partition the file must be repaired or removed locally, + // because older peers do not recognize a delete for a map-missing key. + let capacity = self.consumer_offset_capacity_for(ConsumerKind::ConsumerGroup); let mut dead = Vec::new(); for key in self.consumer_group_offsets.pin().keys() { let Ok(id) = u32::try_from(key.0) else { continue; }; - if !is_live(u64::from(id)) { + if !is_live(u64::from(id)) + && !capacity.is_stranded(id) + && self + .durable_consumer_offsets + .contains(ConsumerKind::ConsumerGroup, id) + { dead.push(id); } } - // A stranded file already failed normal loading or unlink. Reissuing - // replicated deletes every reconciliation pass cannot make its - // filesystem writable and would create a permanent commit loop. - // A single-replica explicit deletion can retry after repair. On a - // replicated partition the file must be repaired or removed locally, - // because older peers do not recognize a delete for a map-missing key. dead.sort_unstable(); dead.dedup(); dead @@ -2449,26 +2466,35 @@ where return; } self.apply_consumer_offset_commit(pending); - if let Err(error) = self.flush_consumer_offset_directories().await { + // Only this request's kind: dirt on the other directory belongs to + // whichever path left it, and its failure is not this client's answer. + // Visibility does not prove crash durability. A failed required barrier + // must remain an error even after the mutation becomes visible. + let mut kinds = [false; 2]; + kinds[crate::state_transfer::consumer_kind_index(kind)] = true; + let failed = self.flush_consumer_offset_directories_for(kinds).await; + if failed.iter().any(|failed| *failed) { if offset.is_some() { self.release_consumer_offset_reservation(kind, consumer_id); } else { - // The delete already took effect in the maps. Stranding keeps - // the key admissible, so the client's retry passes the - // existence check and re-flushes instead of reading 3021. + // Retain retry admission after the visible deletion so a new + // delete can retry its directory barrier instead of returning + // ConsumerOffsetNotFound. self.consumer_offset_capacity_for(kind) .record_stranded(consumer_id); } emit_partition_diag( tracing::Level::WARN, - &PartitionDiagEvent::new(self.diag_ctx(), "no_ack offset directory sync failed") - .with_operation(request_header.operation) - .with_error(error.to_string()), + &PartitionDiagEvent::new( + self.diag_ctx(), + "no_ack offset directory sync failed after a visible local mutation", + ) + .with_operation(request_header.operation), ); Self::send_partition_deny_or_log( &self.consensus, &request_header, - error.as_code(), + IggyError::CannotSyncFile.as_code(), "no_ack offset directory sync failure reply send failed", waiter, ) @@ -2614,6 +2640,14 @@ where kind: ConsumerKind, consumer_id: u32, ) -> Result<(), IggyError> { + if self.durable_consumer_offsets.contains(kind, consumer_id) { + return Ok(()); + } + // A local-only cursor is not a replicated key. Older replicas reject + // missing-key deletes when replaying as primary during a rolling upgrade. + if self.consensus.replica_count() > 1 { + return Err(IggyError::ConsumerOffsetNotFound(consumer_id as usize)); + } let found = match kind { ConsumerKind::Consumer => { let key = usize::try_from(consumer_id).expect("u32 consumer id must fit usize"); @@ -2627,15 +2661,10 @@ where } }; - // The live maps are what `get_consumer_offset` answers from, so a key - // visible there must also be deletable. The durable table covers keys - // a replicated store committed but a NoAck-only apply never mapped. - let durable = self.consensus.replica_count() > 1 - && self.durable_consumer_offsets.contains(kind, consumer_id); let local_stranded = self .consumer_offset_capacity_for(kind) .is_stranded(consumer_id); - if found || durable || local_stranded { + if found || local_stranded { Ok(()) } else { Err(IggyError::ConsumerOffsetNotFound( @@ -2715,7 +2744,9 @@ where let mut rebuilt: HashMap<_, _> = self .pending_consumer_offset_commits .iter() - .filter(|(op, _)| **op >= from_op && (**op <= commit_max || same_view)) + .filter(|(op, _)| { + **op >= from_op && (**op <= commit_max || (same_view && **op <= to_op)) + }) .map(|(op, pending)| (*op, *pending)) .collect(); let headers = self.log.journal().inner.repair_headers_in(from_op..=to_op); @@ -2824,11 +2855,17 @@ where && map.remove(key).is_some() { capacity.forget_inactive_provisional(id); - debug!( + // This cursor was never durable, so the consumer's next `Next` + // poll restarts from offset 0 and redelivers. At-least-once + // permits it; an operator should still see it happen. + warn!( + target: "iggy.partitions.diag", + plane = "partitions", namespace_raw = self.namespace().inner(), ?kind, consumer_id = id, - "reclaimed local consumer offset cursor" + "reclaimed a local consumer offset cursor with no durable backing; its next \ + poll restarts from offset 0" ); remaining -= 1; if remaining == 0 { @@ -3672,10 +3709,15 @@ where /// # Panics /// On mid-iteration status flip. Reachable only if `clear_request_queue` /// is bypassed at view-change reset. - #[allow(clippy::future_not_send)] + #[allow(clippy::future_not_send, clippy::too_many_lines)] pub async fn drain_request_queue_into_prepares(&mut self, slots_freed: usize) { self.resynchronize_consumer_offset_reservations(); let mut promoted = 0; + // Denials do not consume a slot, so without a budget one drain could + // answer the whole queue, each with an awaited reply send, inside one + // commit turn. Consecutive denials end the drain. The shard tick + // resumes parked work even if no further operation commits. + let mut consecutive_denials = 0usize; while promoted < slots_freed { let req = self.consensus().pop_queued_request(); let Some(mut req) = req else { break }; @@ -3722,6 +3764,10 @@ where reply_sender.take(), ) .await; + consecutive_denials += 1; + if consecutive_denials >= PROMOTION_DENIALS_MAX { + break; + } continue; }; if let Some(error) = self.store_offset_range_error(offset) { @@ -3733,6 +3779,10 @@ where reply_sender.take(), ) .await; + consecutive_denials += 1; + if consecutive_denials >= PROMOTION_DENIALS_MAX { + break; + } continue; } if !self @@ -3744,9 +3794,14 @@ where ) .await { + consecutive_denials += 1; + if consecutive_denials >= PROMOTION_DENIALS_MAX { + break; + } continue; } } + consecutive_denials = 0; let prepare = { let consensus = self.consensus(); @@ -3781,6 +3836,23 @@ where } } + #[must_use] + pub fn queued_requests_ready(&self) -> bool { + self.fatal.is_none() + && self.consensus.is_primary() + && self.consensus.is_normal() + && !self.consensus.is_transferring() + && !self.consensus.pipeline_is_full() + && self.consensus.request_queue_len() > 0 + } + + /// Resume a bounded promotion turn without waiting for another commit. + pub async fn resume_queued_requests(&mut self) { + if self.queued_requests_ready() { + self.drain_request_queue_into_prepares(1).await; + } + } + /// # Panics /// Panics on a primary when a prepare's op is ahead of the local /// sequencer: journaling it would make the next op assignment collide, @@ -4547,6 +4619,9 @@ where } } self.consensus.invalidate_local_dvc_suffix(); + let commit_max = self.consensus.commit_max(); + self.pending_consumer_offset_commits + .retain(|op, _| *op < from_op || *op <= commit_max); self.offset_reservations_need_resync.set(true); Ok(removed) } @@ -4932,6 +5007,9 @@ where // commit_min and their replies and dedup folds stay owned by `drained`. // A failure at entry K fences the replica. Recovery replays the whole // unadvanced prefix idempotently, including entries applied before K. + for cell in &self.consumer_offset_dirs_touched { + cell.set(None); + } for (entry, batch_stats) in drained.iter().zip(&committed_batch_stats) { let prepare_header = entry.header; if !self @@ -4982,19 +5060,42 @@ where } // Commit replies and the applied frontier must follow directory - // durability. One sync per touched kind covers the whole walk. A sync - // failure is attributed to the batch boundary because one directory - // sync covers every delete in that kind, not one uniquely failing op. - if let Err(error) = self.flush_consumer_offset_directories().await { - if let Some(entry) = drained.last() { - error!(namespace_raw, %error, "consumer offset directory sync failed after committed operations"); + // durability. One sync per dirty kind covers the whole walk. A sync + // failure is attributed to an operation of that kind in this walk. + // Covered stores depend on previously dirty directory entries too. + // Only a kind THIS walk uses can fence it: dirt left by a NoAck + // request belongs to that request's kind, and fencing a walk that wrote + // consumer offsets over the groups directory would take the node down + // for a failure none of its ops caused. + let touched = [ + self.consumer_offset_dirs_touched[0].replace(None), + self.consumer_offset_dirs_touched[1].replace(None), + ]; + let failed = self + .flush_consumer_offset_directories_for(touched.map(|entry| entry.is_some())) + .await; + if failed.iter().any(|failed| *failed) { + let failed_entry = failed + .iter() + .zip(touched) + .find_map(|(failed, entry)| if *failed { entry } else { None }); + error!( + namespace_raw, + failed_consumer = failed[0], + failed_consumer_group = failed[1], + touched_consumer = touched[0].is_some(), + touched_consumer_group = touched[1].is_some(), + fences = failed_entry.is_some(), + "consumer offset directory sync failed after committed operations" + ); + if let Some((op, operation)) = failed_entry { self.fatal = Some(FatalCommit { namespace_raw, - op: entry.header.op, - operation: entry.header.operation, + op, + operation, }); + return; } - return; } for (mut entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) { @@ -5108,8 +5209,11 @@ where self.drain_request_queue_into_prepares(drained_count).await; } - async fn flush_consumer_offset_directories(&self) -> Result<(), IggyError> { - let mut failed = false; + /// Sync the dirty offset directories selected by `kinds`, indexed by + /// `consumer_kind_index`. Returns which of them failed. A failed directory + /// stays dirty for the next attempt. + async fn flush_consumer_offset_directories_for(&self, kinds: [bool; 2]) -> [bool; 2] { + let mut failed = [false; 2]; for (index, dir) in [ self.consumer_offsets_path.as_deref(), self.consumer_group_offsets_path.as_deref(), @@ -5117,7 +5221,12 @@ where .into_iter() .enumerate() { - if !self.consumer_offset_dirs_dirty[index].get() { + if !kinds[index] || !self.consumer_offset_dirs_dirty[index].get() { + continue; + } + #[cfg(test)] + if self.consumer_offset_dir_sync_fault.get() == Some(index) { + failed[index] = true; continue; } if let Some(dir) = dir { @@ -5128,29 +5237,6 @@ where // There is no remaining dirent whose durability needs // proving. } - Err(error) - if matches!( - error.kind(), - std::io::ErrorKind::InvalidInput | std::io::ErrorKind::Unsupported - ) => - { - // The filesystem refuses directory fsync outright. Nothing - // this replica does will make it succeed, so failing the - // commit walk over it would fence every partition on such - // a mount. - if !self.consumer_offset_dir_sync_unsupported.replace(true) { - warn!( - target: "iggy.partitions.diag", - plane = "partitions", - replica_id = self.consensus.replica(), - namespace_raw = self.namespace().inner(), - path = dir, - error_kind = ?error.kind(), - "filesystem does not support directory fsync; consumer offset \ - unlinks are not crash-durable here" - ); - } - } Err(error) => { warn!( target: "iggy.partitions.diag", @@ -5162,7 +5248,7 @@ where %error, "consumer offset directory sync failed" ); - failed = true; + failed[index] = true; continue; } } @@ -5172,11 +5258,7 @@ where } self.consumer_offset_dirs_dirty[index].set(false); } - if failed { - Err(IggyError::CannotSyncFile) - } else { - Ok(()) - } + failed } fn mark_consumer_offset_dir_dirty(&self, kind: ConsumerKind) { @@ -5184,6 +5266,12 @@ where self.consumer_offset_dirs_dirty[index].set(true); } + /// Keys of `kind` whose offset file could not be loaded or unlinked. + #[must_use] + pub fn stranded_consumer_offset_count(&self, kind: ConsumerKind) -> usize { + self.consumer_offset_capacity_for(kind).stranded_count() + } + /// Batch stats for each drained entry, positionally parallel to `drained`. /// Every entry contributes exactly one slot (`None` for the operations that /// carry no batch), which is what makes the pairing correct by @@ -7155,6 +7243,11 @@ fn accumulate_committed_info( /// it -- only a backlog does, and that one drains over several passes. const SEGMENT_REMOVAL_BUDGET_PER_PASS: usize = 16; +/// Consecutive denials that end one promotion drain. Denials free no slot, so +/// without this bound a single commit turn could answer every queued request, +/// each with an awaited reply send. +const PROMOTION_DENIALS_MAX: usize = 4; + /// What one call to [`IggyPartition::remove_sealed_segments_up_to`] reclaimed. /// /// `budget_spent` reports that the pass stopped on @@ -8683,7 +8776,12 @@ mod tests { partition.consumer_group_offsets_path = Some(dir.path().to_string_lossy().into_owned()); partition.mark_consumer_offset_dir_dirty(ConsumerKind::Consumer); partition.mark_consumer_offset_dir_dirty(ConsumerKind::ConsumerGroup); - assert!(partition.flush_consumer_offset_directories().await.is_err()); + assert_eq!( + partition + .flush_consumer_offset_directories_for([true, true]) + .await, + [true, false] + ); assert!(partition.consumer_offset_dirs_dirty[0].get()); assert!(!partition.consumer_offset_dirs_dirty[1].get()); assert_eq!(partition.offset_dir_sync_count.get(), 1); @@ -8837,6 +8935,100 @@ mod tests { assert!(partition.consensus.pop_queued_request().is_none()); } + #[compio::test] + async fn given_same_view_truncation_when_resynchronizing_should_release_discarded_reservations() + { + let mut partition = test_partition(); + journal_prepare(&partition, 1, Operation::StoreConsumerOffset).await; + journal_prepare(&partition, 2, Operation::StoreConsumerOffset).await; + partition.consensus.sequencer().set_sequence(2); + partition.stage_consumer_offset_upsert(1, ConsumerKind::Consumer, 7, 0, false); + partition.stage_consumer_offset_upsert(2, ConsumerKind::Consumer, 8, 0, false); + partition.truncate_uncommitted_from(2).await.unwrap(); + partition.resynchronize_consumer_offset_reservations(); + assert!(partition.pending_consumer_offset_commits.contains_key(&1)); + assert!(!partition.pending_consumer_offset_commits.contains_key(&2)); + assert_eq!( + partition.occupied_consumer_offset_count(ConsumerKind::Consumer), + 1 + ); + assert!(!partition.consumer_offset_capacity.is_uncertain()); + } + + #[compio::test] + async fn given_promotion_denial_budget_when_no_more_commits_arrive_should_resume_queued_work() { + let (mut partition, sent) = recording_partition_at(0, 3); + partition.set_consumer_offsets_max(1); + partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 0, 0); + partition.consumer_offsets.pin().insert( + 7, + ConsumerOffset::new(ConsumerKind::Consumer, 7, 0, String::new()), + ); + partition.stats.increment_messages_count(1); + for (request, id) in [8, 9, 10, 11, 12, 7].into_iter().enumerate() { + partition + .consensus + .push_queued_request(consensus::RequestEntry::with_sender( + store_offset_request( + 42, + request as u64 + 1, + ConsumerKind::Consumer, + id, + 0, + AckLevel::Quorum, + ), + None, + )) + .unwrap(); + } + partition.drain_request_queue_into_prepares(1).await; + assert_eq!(sent.borrow().len(), PROMOTION_DENIALS_MAX); + assert_eq!(partition.consensus.request_queue_len(), 2); + assert_eq!(partition.consensus.pipeline_len(), 0); + assert!(partition.queued_requests_ready()); + partition.resume_queued_requests().await; + assert_eq!(partition.consensus.request_queue_len(), 0); + assert_eq!(partition.consensus.pipeline_len(), 1); + assert!(!partition.queued_requests_ready()); + } + + #[compio::test] + async fn given_covered_store_with_dirty_directory_when_sync_fails_should_fence_its_own_operation() + { + let dir = tempfile::tempdir().unwrap(); + let (mut partition, sent) = recording_partition_at(0, 3); + partition.consumer_offsets_path = Some(dir.path().to_string_lossy().into_owned()); + partition.consumer_offset_enforce_fsync = true; + let pending = + PendingConsumerOffsetCommit::upsert_auto_commit(ConsumerKind::Consumer, 7, 10); + partition + .persist_consumer_offset_commit(pending) + .await + .unwrap(); + partition.apply_consumer_offset_commit(pending); + partition.consumer_offset_dir_sync_fault.set(Some(0)); + partition.stage_consumer_offset_upsert(1, ConsumerKind::Consumer, 7, 5, true); + partition.consensus.restore_commit_state(0, 1); + let header = PrepareHeader { + op: 1, + operation: Operation::StoreConsumerOffset, + client: 42, + request: 1, + ..Default::default() + }; + partition + .handle_committed_entries(vec![PipelineEntry::new(header)], &repair_config(), true) + .await; + assert_eq!(partition.fatal.as_ref().unwrap().op, 1); + assert_eq!( + partition.fatal.as_ref().unwrap().operation, + Operation::StoreConsumerOffset + ); + assert_eq!(partition.consensus.commit_min(), 0); + assert!(sent.borrow().is_empty()); + assert!(partition.consumer_offset_dirs_dirty[0].get()); + } + #[compio::test] async fn given_repeated_admitted_deletes_when_committed_should_be_idempotent() { let (mut partition, _) = recording_partition(); @@ -9623,14 +9815,31 @@ mod tests { std::fs::create_dir_all(group_dir.join("7")).unwrap(); let (mut partition, _) = recording_partition(); partition.consumer_group_offsets_path = Some(group_dir.to_string_lossy().into_owned()); - partition.seed_stranded_consumer_offset(ConsumerKind::ConsumerGroup, 7); + // In the live map, so the reclaim walk actually meets the key and the + // stranded filter is what keeps it out of the delete log. + partition.seed_recovered_consumer_offset(ConsumerKind::ConsumerGroup, 7, 11, 11); + partition.consumer_group_offsets.pin().insert( + ConsumerGroupId(7), + ConsumerOffset::new(ConsumerKind::ConsumerGroup, 7, 11, String::new()), + ); + partition.consumer_group_offset_capacity.record_stranded(7); assert!(partition.consumer_group_offset_capacity.is_stranded(7)); + assert_eq!( + partition.dead_consumer_group_offset_ids(|_| true), + Vec::<u32>::new() + ); assert!( partition .dead_consumer_group_offset_ids(|_| false) .is_empty(), "automatic cleanup must not resubmit a known failed unlink" ); + partition.consumer_group_offset_capacity.clear_stranded(7); + assert_eq!( + partition.dead_consumer_group_offset_ids(|_| false), + vec![7], + "the same dead key is reclaimed once it is no longer stranded" + ); assert!( partition .offsets_wire_snapshot_for_test() @@ -9640,17 +9849,14 @@ mod tests { } #[test] - fn given_visible_or_stranded_key_when_replicated_should_admit_the_delete() { - // Whatever `get_consumer_offset` can answer from, or whatever a file on - // disk may resurrect at boot, must be deletable. A replicated delete of a - // key some replica never held applies there as a no-op. + fn given_local_only_key_when_replicated_should_require_durable_membership_to_delete() { let (partition, _) = recording_partition_at(0, 3); partition.seed_stranded_consumer_offset(ConsumerKind::ConsumerGroup, 7); assert!( partition .ensure_consumer_offset_exists(ConsumerKind::ConsumerGroup, 7) - .is_ok(), - "a stranded key is exactly what a delete settles" + .is_err(), + "a stranded file is not a replicated key" ); partition.consumer_group_offsets.pin().insert( ConsumerGroupId(8), @@ -9659,8 +9865,8 @@ mod tests { assert!( partition .ensure_consumer_offset_exists(ConsumerKind::ConsumerGroup, 8) - .is_ok(), - "a key a read can see must be deletable" + .is_err(), + "a local poll cursor must not produce a missing-key delete on an older peer" ); partition .durable_consumer_offsets @@ -9687,7 +9893,7 @@ mod tests { } #[compio::test] - async fn given_committed_delete_when_unlink_fails_should_apply_and_strand_the_key() { + async fn given_committed_delete_when_unlink_fails_should_preserve_state_and_report_failure() { let dir = tempfile::tempdir().unwrap(); let group_dir = dir.path().join("groups"); std::fs::create_dir_all(group_dir.join("7")).unwrap(); @@ -9700,21 +9906,19 @@ mod tests { partition.seed_recovered_consumer_offset(ConsumerKind::ConsumerGroup, 7, 11, 11); partition.stage_consumer_offset_delete(1, ConsumerKind::ConsumerGroup, 7); - // The committed delete applies on every replica regardless of the - // unlink. The key stays stranded so the directory entry that may - // resurrect at boot is still counted and still deletable. + // A failed file mutation cannot become a successful logical delete. assert!( partition .apply_staged_consumer_offset_commit(1) .await - .is_ok() + .is_err() ); - assert!(partition.consumer_group_offset_ids().is_empty()); + assert_eq!(partition.consumer_group_offset_ids(), vec![7]); assert_eq!( partition.durable_consumer_offset_count(ConsumerKind::ConsumerGroup), - 0 + 1 ); - assert!(!partition.pending_consumer_offset_commits.contains_key(&1)); + assert!(partition.pending_consumer_offset_commits.contains_key(&1)); assert!(partition.consumer_group_offset_capacity.is_stranded(7)); assert!(partition.fatal.is_none()); } @@ -9761,18 +9965,17 @@ mod tests { } #[compio::test] - async fn given_no_ack_store_when_genuine_directory_sync_fails_should_keep_maps_consistent() { + async fn given_no_ack_store_when_its_directory_sync_fails_should_report_failure_and_stay_dirty() + { let dir = tempfile::tempdir().unwrap(); let (mut partition, sent) = recording_partition(); partition.consumer_offsets_path = Some(dir.path().join("consumers").to_string_lossy().into_owned()); - // A path under a regular file: opening it fails with ENOTDIR, which is - // neither the tolerated "gone" nor the tolerated "unsupported". - let not_a_dir = dir.path().join("file"); - std::fs::write(¬_a_dir, b"not a directory").unwrap(); partition.consumer_group_offsets_path = - Some(not_a_dir.join("groups").to_string_lossy().into_owned()); - partition.consumer_offset_dirs_dirty[1].set(true); + Some(dir.path().join("groups").to_string_lossy().into_owned()); + partition.consumer_offset_enforce_fsync = true; + // Visibility alone cannot satisfy the explicitly requested barrier. + partition.consumer_offset_dir_sync_fault.set(Some(0)); partition.stats.increment_messages_count(1); partition @@ -9797,8 +10000,133 @@ mod tests { ) .ok() }) - .expect("sync failure reply"); + .expect("failure reply"); assert_eq!(header.status, IggyError::CannotSyncFile.as_code()); + assert!( + partition.consumer_offset_dirs_dirty[0].get(), + "the failed directory keeps its dirt for the next walk" + ); + } + + #[compio::test] + async fn given_stale_dirt_on_the_other_kind_when_its_sync_fails_should_not_fence_the_walk() { + let dir = tempfile::tempdir().unwrap(); + let (mut partition, sent) = recording_partition_at(0, 3); + partition.consumer_offsets_path = + Some(dir.path().join("consumers").to_string_lossy().into_owned()); + partition.consumer_group_offsets_path = + Some(dir.path().join("groups").to_string_lossy().into_owned()); + partition.consumer_offset_enforce_fsync = true; + // Dirt a NoAck request left on the groups directory, whose sync now + // fails. This walk writes consumer offsets only. + partition.consumer_offset_dirs_dirty[1].set(true); + partition.consumer_offset_dir_sync_fault.set(Some(1)); + partition + .persist_consumer_offset_commit(PendingConsumerOffsetCommit::upsert( + ConsumerKind::Consumer, + 1, + 0, + )) + .await + .unwrap(); + partition.stage_consumer_offset_delete(1, ConsumerKind::Consumer, 1); + let header = PrepareHeader { + op: 1, + operation: Operation::DeleteConsumerOffset, + client: 42, + request: 1, + ..Default::default() + }; + partition.consensus.restore_commit_state(0, 1); + partition + .handle_committed_entries(vec![PipelineEntry::new(header)], &repair_config(), true) + .await; + assert!( + partition.fatal.is_none(), + "a failure on a kind this walk did not write must not fence it" + ); + assert_eq!(partition.consensus.commit_min(), 1); + assert_eq!(sent.borrow().len(), 1); + assert!(!partition.consumer_offset_dirs_dirty[0].get()); + assert!(partition.consumer_offset_dirs_dirty[1].get()); + + // The same failure on the kind the walk wrote fences it. + partition.consumer_offset_dir_sync_fault.set(Some(0)); + partition + .persist_consumer_offset_commit(PendingConsumerOffsetCommit::upsert( + ConsumerKind::Consumer, + 2, + 0, + )) + .await + .unwrap(); + partition.stage_consumer_offset_delete(2, ConsumerKind::Consumer, 2); + let header = PrepareHeader { + op: 2, + operation: Operation::DeleteConsumerOffset, + client: 42, + request: 2, + ..Default::default() + }; + partition.consensus.advance_commit_max(2); + partition + .handle_committed_entries(vec![PipelineEntry::new(header)], &repair_config(), true) + .await; + assert!( + partition.fatal.is_some(), + "a sync failure on a written kind fences the walk" + ); + assert_eq!(partition.consensus.commit_min(), 1); + } + + #[compio::test] + async fn given_no_ack_delete_sync_failure_when_retried_should_retry_barrier_and_succeed() { + let dir = tempfile::tempdir().unwrap(); + let (mut partition, sent) = recording_partition(); + partition.consumer_offsets_path = Some(dir.path().to_string_lossy().into_owned()); + partition.consumer_offset_enforce_fsync = true; + let stored = PendingConsumerOffsetCommit::upsert(ConsumerKind::Consumer, 7, 0); + partition + .persist_consumer_offset_commit(stored) + .await + .unwrap(); + partition.apply_consumer_offset_commit(stored); + partition.consumer_offset_dir_sync_fault.set(Some(0)); + partition + .apply_consumer_offset_no_ack( + Box::new(*delete_offset_request(42, 1, 7).header()), + ConsumerKind::Consumer, + 7, + None, + None, + ) + .await; + let status = |index: usize| { + let frames = sent.borrow(); + let bytes = frames[index].1.as_slice(); + let start = std::mem::offset_of!(ReplyHeader, status); + u32::from_le_bytes(bytes[start..start + 4].try_into().unwrap()) + }; + assert_eq!(status(0), IggyError::CannotSyncFile.as_code()); + assert!(!dir.path().join("7").exists()); + assert!( + partition + .ensure_consumer_offset_exists(ConsumerKind::Consumer, 7) + .is_ok() + ); + partition.consumer_offset_dir_sync_fault.set(None); + partition + .apply_consumer_offset_no_ack( + Box::new(*delete_offset_request(42, 2, 7).header()), + ConsumerKind::Consumer, + 7, + None, + None, + ) + .await; + assert_eq!(status(1), 0); + assert!(!partition.consumer_offset_dirs_dirty[0].get()); + assert!(!partition.consumer_offset_capacity.is_stranded(7)); } #[test] diff --git a/core/partitions/src/offset_storage.rs b/core/partitions/src/offset_storage.rs index 605ea12d0..f750b1345 100644 --- a/core/partitions/src/offset_storage.rs +++ b/core/partitions/src/offset_storage.rs @@ -51,7 +51,9 @@ pub enum OffsetRecord { /// A usable offset. `checksummed` is false for a bare offset predating the /// checksum, read as-is and upgraded by the next write. Value { offset: u64, checksummed: bool }, - /// Shorter than the value, such as a legacy interrupted in-place write. + /// Shorter than the value: a crash between the truncate and the write of an + /// in-place update, the default path while `consumer_offset_enforce_fsync` + /// is off. Torn, /// The checksum does not describe the value stored beside it. Corrupt { @@ -109,12 +111,13 @@ pub fn decode_offset_record(bytes: &[u8]) -> OffsetRecord { /// Overwrite a consumer-offset file with `offset` and a checksum over it. /// -/// Without `enforce_fsync` the file is rewritten in place. With it, the record -/// goes to a sibling inode, is data-synced and renamed over the prior file, so a -/// failed write leaves the prior cursor intact. The replacement is tied to the -/// same knob as the sync: without the sync neither the write nor the rename is -/// ordered against a crash, so the extra inode and rename buy nothing. The -/// caller syncs the parent directory afterwards. +/// Without `enforce_fsync` the file is rewritten in place and no directory is +/// synced. With it, the record goes to a sibling inode, is data-synced and +/// renamed over the prior file, so a failed write leaves the prior cursor +/// intact, and the caller marks the parent directory for a sync on the next +/// commit walk. The replacement is tied to the same knob as the sync: without +/// the sync neither the write nor the rename is ordered against a crash, so +/// the extra inode and rename buy nothing. /// /// # Errors /// [`IggyError`] when the directory, file, or write cannot be created or completed. diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs index 29b71f944..6b8a01e00 100644 --- a/core/partitions/src/poll_plan.rs +++ b/core/partitions/src/poll_plan.rs @@ -942,7 +942,7 @@ impl AutoCommitCtx { impl AutoCommitApplied { /// Record the group handoff frontier only after poll admission succeeds. - pub fn mark_served(&self) { + fn mark_served(&self) { if let Some(last_polled) = &self.last_polled { last_polled.record(self.offset); } @@ -971,9 +971,30 @@ impl AutoCommitApplied { Rc::ptr_eq(&self.durable, durable) } - /// Undo this poll's eager update after synchronous admission fails. The - /// caller must not yield between execution and this rollback. - pub fn rollback_created(&self) { + /// Run the serving shard's synchronous admission and settle this apply in + /// the same call: `Ok` marks the cursor served, `Err` rolls the eager + /// local update back and returns the error. The rollback is a bare store of + /// the previous offset, so nothing may yield between the decision and it. + /// Keeping both inside one synchronous method is what makes that hold for + /// every caller. + /// + /// # Errors + /// Whatever `decide` returned, after the rollback. + pub fn admit<E>(self, decide: impl FnOnce(&Self) -> Result<(), E>) -> Result<(), E> { + match decide(&self) { + Ok(()) => { + self.mark_served(); + Ok(()) + } + Err(error) => { + self.rollback_created(); + Err(error) + } + } + } + + /// Undo this poll's eager update after synchronous admission fails. + fn rollback_created(&self) { match &self.target { AutoCommitTarget::Consumer { offsets, @@ -1022,6 +1043,9 @@ fn apply_local_offset<K: Hash + Eq + Clone + Send + Sync>( if let Some(existing) = guard.get(&key) { return Ok(Some(existing.offset.fetch_max(offset, Ordering::Relaxed))); } + // The `len()` read and the insert are not atomic on this lock-free map. + // The bound holds because every poll of one partition runs on that + // partition's own shard thread, so no second inserter exists. capacity.admit_local_map_key(guard.len(), durable_full)?; guard.insert(key, create()); capacity.note_local_key_change(); diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index 72fa5e729..94a62b60e 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -1662,7 +1662,13 @@ async fn retry_offset_mutation<T, E: fmt::Debug, F: Future<Output = Result<T, E> } } } - operation().await + operation().await.inspect_err(|error| { + tracing::debug!( + attempt = OFFSET_IO_ATTEMPTS, + ?error, + "offset mutation failed on the last attempt" + ); + }) } /// One consumer-offset file the install is about to write. Collected before any @@ -1792,7 +1798,6 @@ where if self.repair.is_some() { return Err(PartitionTransferUnavailable::RepairInProgress); } - self.validate_consumer_offset_transfer_counts()?; // Primary-by-index at view 0 over an empty log passes every gate above // yet knows nothing: a group whose directory is absent boots through // `consensus.init()`, comes up Normal at view 0, and an empty group is @@ -2403,7 +2408,8 @@ where offsets_bytes: &[u8], committed_purge_generation: u64, ) -> Result<PartitionInstallOutcome, PartitionInstallError> { - // ---- check phase: nothing below may mutate ---- + // ---- check phase: nothing below may mutate live state. Staging + // writes only sibling files the install can abandon. ---- let Some(partition_dir) = self.partition_dir.clone() else { return Err(PartitionInstallError::NoPartitionDir); }; @@ -2959,10 +2965,8 @@ where .map(|(kind, id, path)| ((kind, id), path)) .collect(); for ((kind, consumer_id), path) in old_paths { - // Strand rather than fail: failing here fences the partition and - // re-pulls the same transfer against the same unlinkable file. A - // stranded key stays counted and admits a later delete, which is - // what the purge path does with the same fault. + // An obsolete authoritative file must not survive a successful + // install because boot would reload it outside the incoming table. match delete_persisted_offset(&path).await { Ok(removed) => { if removed { @@ -2983,6 +2987,10 @@ where %error, "install could not remove a superseded consumer offset file" ); + return Err(PartitionInstallError::OffsetPersistence { + path, + source: error, + }); } } } @@ -3225,8 +3233,7 @@ where } } OffsetDirEntry::Offset { id, path } => { - // Same policy as install: an unlinkable file strands its - // key instead of failing the converge and looping. + // A remaining authoritative file prevents convergence. match retry_offset_mutation(|| delete_persisted_offset(&path)).await { Ok(_) => self.consumer_offset_capacity_for(kind).clear_stranded(id), Err(error) => { @@ -3237,15 +3244,19 @@ where %error, "converge could not remove a consumer offset file" ); + return Err(error); } } } } } - if std::path::Path::new(dir).exists() { - fsync_dir(dir) - .await - .map_err(|_| iggy_common::IggyError::CannotSyncFile)?; + // No `exists()` probe: that is a blocking stat on the pump. A + // directory the unlinks emptied and removed has nothing left to + // make durable. + match fsync_dir(dir).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(_) => return Err(iggy_common::IggyError::CannotSyncFile), } } // Every segment and staging file this partition had is about to be diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs index 85ecb69b5..e6f66ba49 100644 --- a/core/sdk/src/clients/consumer.rs +++ b/core/sdk/src/clients/consumer.rs @@ -468,8 +468,9 @@ unsafe impl Sync for IggyConsumer {} /// /// A failed poll request waits [`polling_retry_interval()`] before yielding `Some(Err(..))` and leaves /// the consumer usable for the next call. This delay also applies to terminal server errors and when -/// the ordinary poll interval is disabled. Connection and authentication failures pause polling until -/// the client has reconnected and signed in again, which the consumer handles automatically. Hence, +/// the ordinary poll interval is disabled. Connection and authentication failures are yielded at once +/// and pause polling until the client has reconnected and signed in again, which the consumer handles +/// automatically; the next call parks for [`polling_retry_interval()`] while polling is paused. Hence, /// deciding when to give up on repeated errors is up to you. /// /// For a boilerplate implementation of such a loop Iggy provides [`IggyConsumerMessageExt::consume_messages`]. @@ -549,7 +550,7 @@ unsafe impl Sync for IggyConsumer {} /// | [`allow_replay()`] | off | whether a message can be handed over again | /// | [`auto_join_consumer_group()`] | on | joining the group during [`init()`](Self::init) and again whenever the membership is lost. With [`do_not_auto_join_consumer_group()`] joining is up to the caller, and a poll without a membership fails with [`IggyError::ConsumerGroupMemberNotFound`] | /// | [`create_consumer_group_if_not_exists()`] | on | creating the group when it is missing | -/// | [`polling_retry_interval()`] | one second | delay before yielding a poll error, and between attempts while polling is blocked or the member holds no partitions | +/// | [`polling_retry_interval()`] | one second | delay before yielding poll errors other than connection and authentication failures, and between attempts while polling is blocked or the member holds no partitions | /// | [`init_retries()`] | none, one second apart | retries when the stream or topic is missing at [`init()`](Self::init) | /// | [`offset_drain_timeout()`] | five seconds | how long [`shutdown()`](Self::shutdown) waits for pending commits | /// | [`encryptor()`] | inherited from the client | decrypting payloads and user headers, see [Encryption](#encryption) | @@ -1360,8 +1361,11 @@ impl IggyConsumer { return Ok(PolledMessages::empty()); } - // Handle connection/auth errors - disable polling until event task re-enables - // it after reconnection and rejoin complete + // Connection and auth errors: disable polling until the event task + // re-enables it after reconnection and rejoin complete. Yielded at + // once: the next poll already parks on `can_poll`, so a retry sleep + // here would only delay the caller's view of an error it does not + // act on. if matches!( error, IggyError::Disconnected | IggyError::Unauthenticated | IggyError::StaleClient @@ -1370,6 +1374,7 @@ impl IggyConsumer { if is_consumer_group { joined_consumer_group.store(false, ORDERING); } + return Err(error); } trace!("Retrying to poll messages in {retry_interval}..."); sleep(retry_interval.get_duration()).await; diff --git a/core/sdk/src/leader_aware.rs b/core/sdk/src/leader_aware.rs index 77da57095..0319451e2 100644 --- a/core/sdk/src/leader_aware.rs +++ b/core/sdk/src/leader_aware.rs @@ -355,6 +355,8 @@ fn normalize_address(addr: &str) -> String { /// failed dial cannot cycle the request back through nodes it already tried. #[derive(Debug)] pub(crate) struct RosterWalk { + /// Whether the roster named at least one node when the walk was built. + roster_known: bool, remaining: VecDeque<String>, attempted: Vec<String>, } @@ -382,6 +384,7 @@ impl RosterWalk { Self { remaining: ordered, attempted: vec![current.to_owned()], + roster_known: !roster.is_empty(), } } @@ -407,8 +410,11 @@ impl RosterWalk { Some(endpoint) } + /// True only when the roster itself names one node. An empty roster (its + /// discovery failed) also leaves nothing to walk, but replaying that one + /// address would be a guess, not a decision. pub(crate) fn is_single_endpoint(&self) -> bool { - self.remaining.is_empty() && self.attempted.len() == 1 + self.roster_known && self.remaining.is_empty() && self.attempted.len() == 1 } } @@ -794,6 +800,7 @@ mod tests { #[test] fn given_single_endpoint_roster_when_exhausted_should_allow_local_retry() { + assert!(!RosterWalk::new("127.0.0.1:8090", &[]).is_single_endpoint()); let mut walk = RosterWalk::new("127.0.0.1:8090", &["localhost:8090".to_owned()]); assert!(walk.is_single_endpoint()); assert_eq!(walk.next(), None); diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index d65448819..b559f81ba 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -1280,6 +1280,17 @@ impl TcpClient { return Err(IggyError::TransientNotAccepted); } + if roster_walk + .as_ref() + .is_some_and(RosterWalk::is_single_endpoint) + { + // Discovery already proved there is no routing choice. + // Retry without reacquiring the lock so reconnects and + // unrelated requests do not wait out this deadline. + drop(routing_guard.take()); + continue; + } + if routing_guard.is_none() { routing_guard = Some( tokio::time::timeout_at(overall_deadline, self.routing_lock.lock()) @@ -1332,8 +1343,9 @@ impl TcpClient { .is_some_and(RosterWalk::is_single_endpoint) { // Partition materialisation can outlast the short retry - // window on a single node. No routing change is needed, - // so let other requests and reconnects acquire the lock. + // window on a single node. No routing change is needed. + // Subsequent refusals take the lock-free local retry + // branch above rather than reacquiring this guard. drop(routing_guard.take()); continue; } else { diff --git a/core/server/config.toml b/core/server/config.toml index 06b0af125..3cc021375 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -1028,8 +1028,11 @@ dedup_clients_max = 4096 # Standalone consumers and consumer groups are counted separately. Existing # keys remain writable at the limit. A new key is rejected before consensus. # A non-empty auto_commit poll that needs a new offset key is also rejected -# with TooManyConsumerOffsets and returns no messages. Polls with auto_commit -# disabled do not allocate offset keys and remain available. +# with TooManyConsumerOffsets and returns no messages. A poll whose auto_commit +# cannot be submitted (the owning shard's inbox is full, or the partition +# changed primary during the read) is rejected with TransientNotAccepted, also +# without messages; both are retriable. Polls with auto_commit disabled do not +# allocate offset keys and remain available. # UPGRADE: existing files are loaded even above this limit, but new keys then # remain blocked until offsets are explicitly deleted or this limit is raised. # Standalone offsets have no automatic expiry. Reuse stable consumer ids and @@ -1044,10 +1047,10 @@ consumer_offsets_max = 4096 # offset store writes a sibling file, fdatasyncs it and renames it over the # prior cursor, and the offsets directory is fsynced once per commit walk before # any reply in that walk is sent. Independent of the topic's enforce_fsync, -# which governs message and index files: a replicated partition -# already gets its durability from the quorum, and a lost offset file costs at -# most a redelivery from the last flushed cursor. Off, the file is rewritten in -# place with no sync. +# which governs message and index files. Off, the file is rewritten in place +# with no sync. A lost or torn cursor can cause replay from the earliest retained +# data. On, the filesystem must support both file and directory sync. A failed +# required barrier is reported as a failure, even if the mutation is visible. # The environment override is IGGY_PARTITION_CONSUMER_OFFSET_ENFORCE_FSYNC. consumer_offset_enforce_fsync = false diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs index f7527895c..ff6f4f16f 100644 --- a/core/server/src/boot/recovery.rs +++ b/core/server/src/boot/recovery.rs @@ -338,7 +338,9 @@ const _: () = assert!( ); const _: () = assert!( 4 * configs::partition::PARTITION_CONSUMER_OFFSETS_CEILING - <= partitions::CONSUMER_OFFSETS_ENTRIES_MAX as usize + <= partitions::CONSUMER_OFFSETS_ENTRIES_MAX as usize, + "four ceilings fill the transfer decoder's entry budget exactly, with no headroom left; raise \ + CONSUMER_OFFSETS_ENTRIES_MAX before raising PARTITION_CONSUMER_OFFSETS_CEILING" ); const _: () = assert!(configs::metadata::DEFAULT_METADATA_CLIENTS_TABLE_MAX == consensus::CLIENTS_TABLE_MAX); diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index a25529cbb..76fcd985c 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -216,7 +216,7 @@ where match result { Ok((fragments, current_offset, auto_commit)) => { if let Some(applied) = auto_commit - && let Err(error) = submit_auto_commit(shard, namespace, &applied) + && let Err(error) = submit_auto_commit(shard, namespace, applied) { PartitionReadReply::Rejected(error) } else { @@ -246,7 +246,17 @@ where /// poll is served on whichever node owns the namespace locally, which may be a /// backup. So gate on primary status here. Auto-commit /// is server-managed best-effort (at-least-once delivery), so a follower-served -/// poll simply does not advance the durable offset. +/// poll simply does not advance the durable offset. The same contract covers a +/// local cursor that never became durable: when the per-kind live map is over +/// its limit the partition evicts such a cursor, and that consumer's next +/// `Next` poll restarts from offset 0. +/// +/// A poll whose auto-commit cannot be submitted is answered +/// `TransientNotAccepted` and returns no messages, even though the fragments +/// were already read: the owning shard's inbox refused the frame, or the +/// partition changed primary or incarnation during the read. Like the +/// `TooManyConsumerOffsets` refusal at the key limit, it returns no batch. +/// Transient refusals permit retry. Capacity refusals need available capacity. /// /// Coalescing: an offset the partition's committed high-water already covers is /// dropped without a consensus op (the steady state for a re-poll of committed @@ -256,7 +266,7 @@ where fn submit_auto_commit<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, namespace: IggyNamespace, - applied: &AutoCommitApplied, + applied: AutoCommitApplied, ) -> Result<(), IggyError> where B: ShellBus, @@ -265,66 +275,55 @@ where S: 'static, SB: SuperblockStore + 'static, { - let primary = shard - .plane - .partitions() - .with_partition(&namespace, |partition| { - let consensus = partition.consensus(); - if !partition.auto_commit_admission_ready(applied) { - return Err(IggyError::TransientNotAccepted); - } - Ok(consensus.is_primary() && consensus.is_normal() && !consensus.is_transferring()) - }); - if matches!(primary, None | Some(Err(_))) { - applied.rollback_created(); - return Err(IggyError::TransientNotAccepted); - } - if primary == Some(Ok(false)) { - applied.mark_served(); - debug!( - namespace_raw = namespace.inner(), - "auto-commit offset not replicated: partition not primary on this node (best-effort)" - ); - return Ok(()); - } - let reservation = match applied.reserve_durable() { - Ok(Some(reservation)) => reservation, - Ok(None) => { - applied.mark_served(); + // Everything inside is synchronous: `admit` rolls the eager cursor update + // back on `Err` in the same call, and no await may sit between the update + // and that rollback. + applied.admit(|applied| { + let primary = shard + .plane + .partitions() + .with_partition(&namespace, |partition| { + let consensus = partition.consensus(); + if !partition.auto_commit_admission_ready(applied) { + return Err(IggyError::TransientNotAccepted); + } + Ok(consensus.is_primary() && consensus.is_normal() && !consensus.is_transferring()) + }); + if matches!(primary, None | Some(Err(_))) { + return Err(IggyError::TransientNotAccepted); + } + if primary == Some(Ok(false)) { + debug!( + namespace_raw = namespace.inner(), + "auto-commit offset not replicated: partition not primary on this node (best-effort)" + ); return Ok(()); } - Err(error) => { - if !error.uncertain { - shard.metrics().record_consumer_offset_denied(applied.kind); + let reservation = match applied.reserve_durable() { + Ok(Some(reservation)) => reservation, + Ok(None) => return Ok(()), + Err(error) => { + if !error.uncertain { + shard.metrics().record_consumer_offset_denied(applied.kind); + } + warn_auto_commit_capacity(namespace, error); + return Err(error.into()); } - warn_auto_commit_capacity(namespace, error); - applied.rollback_created(); - return Err(error.into()); - } - }; - let message = match build_auto_commit_request(namespace, applied) { - Ok(message) => message, - Err(error) => { - applied.rollback_created(); + }; + let message = build_auto_commit_request(namespace, applied).inspect_err(|error| { warn!( namespace_raw = namespace.inner(), error = %error, "failed to build auto-commit store-offset request" ); - return Err(error); - } - }; - // Routes by namespace to this same owning primary shard's inbox. The pump - // admits it next turn exactly like a client store. `dispatch` never blocks. - if shard - .submit_auto_commit_offset(message, reservation) - .is_err() - { - applied.rollback_created(); - return Err(IggyError::TransientNotAccepted); - } - applied.mark_served(); - Ok(()) + })?; + // Routes by namespace to this same owning primary shard's inbox. The + // pump admits it next turn exactly like a client store. `dispatch` + // never blocks. + shard + .submit_auto_commit_offset(message, reservation) + .map_err(|_| IggyError::TransientNotAccepted) + }) } fn warn_auto_commit_capacity( diff --git a/core/server/src/offset_recovery.rs b/core/server/src/offset_recovery.rs index 99f64dc30..40dca3ff8 100644 --- a/core/server/src/offset_recovery.rs +++ b/core/server/src/offset_recovery.rs @@ -28,10 +28,15 @@ use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyError}; use partitions::offset_storage::{OffsetRecord, decode_offset_record, offset_replacement_id}; +use std::path::PathBuf; use std::sync::atomic::AtomicU64; +use tokio::sync::{Semaphore, mpsc}; use tracing::{error, trace, warn}; const COMPONENT: &str = "STREAMING_PARTITIONS"; +const OFFSET_DIRECTORY_BUFFER: usize = 64; +static OFFSET_DIRECTORY_READERS: Semaphore = Semaphore::const_new(4); +type OffsetDirectoryEntries = mpsc::Receiver<std::io::Result<Option<PathBuf>>>; pub struct RecoveredOffsets<T> { pub entries: Vec<T>, @@ -76,30 +81,25 @@ async fn load_offsets<T>( construct: impl Fn(ConsumerOffset) -> T, ) -> Result<RecoveredOffsets<T>, IggyError> { trace!(?kind, path, "loading consumer offsets"); - let dir_entries = std::fs::read_dir(path) - .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?; + let mut dir_entries = offset_directory_entries(path).await?; let mut recovered = RecoveredOffsets::default(); - for dir_entry in dir_entries { - let dir_entry = match dir_entry { - Ok(entry) => entry, - Err(error) => { - warn!(?kind, path, %error, "failed to read offset directory entry"); - continue; - } - }; - let file_type = match dir_entry.file_type() { - Ok(file_type) => file_type, - Err(error) => { - warn!(?kind, path, %error, "failed to read offset entry type"); - continue; + loop { + let entry_path = match dir_entries.recv().await { + Some(Ok(Some(path))) => path, + Some(Ok(None)) => break, + Some(Err(error)) => { + warn!(?kind, path, %error, "failed to enumerate offset directory"); + return Err(IggyError::CannotReadConsumerOffsets(path.to_owned())); } + None => return Err(IggyError::CannotReadConsumerOffsets(path.to_owned())), }; - if !file_type.is_file() { - continue; - } - let name = dir_entry.file_name().to_string_lossy().into_owned(); + let name = entry_path + .file_name() + .unwrap_or_default() + .to_string_lossy() + .into_owned(); if offset_replacement_id(&name).is_some() { - remove_stale_replacement(&dir_entry.path(), &name).await; + remove_stale_replacement(&entry_path, &name).await; continue; } let Ok(consumer_id) = name.parse::<u32>() else { @@ -109,7 +109,7 @@ async fn load_offsets<T>( ); continue; }; - let Some(path) = dir_entry.path().to_str().map(str::to_owned) else { + let Some(path) = entry_path.to_str().map(str::to_owned) else { error!(?kind, name, "invalid consumer offset path"); continue; }; @@ -131,6 +131,57 @@ async fn load_offsets<T>( Ok(recovered) } +async fn offset_directory_entries(path: &str) -> Result<OffsetDirectoryEntries, IggyError> { + // Compio has no asynchronous directory iterator and the shard's blocking + // pool is disabled. Bound both OS threads and buffered paths. The worker + // owns the permit so cancellation cannot exceed the concurrency bound. + let permit = OFFSET_DIRECTORY_READERS + .acquire() + .await + .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?; + let (sender, receiver) = mpsc::channel(OFFSET_DIRECTORY_BUFFER); + let directory = path.to_owned(); + std::thread::Builder::new() + .name("iggy-offset-recovery".to_owned()) + .spawn(move || { + let _permit = permit; + let result = (|| { + // Only the directory read itself is fatal. One unreadable + // entry is skipped with a warning, as the on-reactor loader + // did, so a single bad dirent cannot keep a partition from + // booting. + for entry in std::fs::read_dir(&directory)? { + let entry = match entry { + Ok(entry) => entry, + Err(error) => { + warn!(path = directory, %error, "failed to read offset directory entry"); + continue; + } + }; + let is_file = match entry.file_type() { + Ok(file_type) => file_type.is_file(), + Err(error) => { + warn!(path = directory, %error, "failed to read offset entry type"); + continue; + } + }; + if is_file && sender.blocking_send(Ok(Some(entry.path()))).is_err() { + return Ok(()); + } + } + Ok(()) + })(); + // Explicit completion distinguishes an empty directory from an + // interrupted worker. Closed receivers simply abandon enumeration. + let _ = sender.blocking_send(result.map(|()| None)); + }) + .map_err(|error| { + error!(path, %error, "failed to start offset directory reader"); + IggyError::CannotReadConsumerOffsets(path.to_owned()) + })?; + Ok(receiver) +} + /// A crashed atomic replacement leaves its sibling behind. The rename never /// landed, so the sibling is never authoritative. Removal needs no directory /// sync because a resurrected sibling is still ignored on the next load. @@ -224,6 +275,31 @@ async fn remove_invalid_offset_file(path: &str, offset_kind: &'static str) -> Of mod tests { use super::*; + #[compio::test] + async fn given_missing_directory_when_loading_should_report_error_instead_of_empty_state() { + let dir = tempfile::tempdir().unwrap(); + let missing = dir.path().join("missing"); + assert!(matches!( + load_consumer_offsets(missing.to_str().unwrap()).await, + Err(IggyError::CannotReadConsumerOffsets(_)) + )); + } + + #[compio::test] + async fn given_full_directory_buffer_when_loader_is_cancelled_should_release_worker_capacity() { + let dir = tempfile::tempdir().unwrap(); + for id in 0..OFFSET_DIRECTORY_BUFFER * 2 { + std::fs::write(dir.path().join(id.to_string()), 0_u64.to_le_bytes()).unwrap(); + } + let path = dir.path().to_str().unwrap(); + for _ in 0..8 { + let entries = offset_directory_entries(path).await.unwrap(); + drop(entries); + } + let loaded = load_consumer_offsets(path).await.unwrap(); + assert_eq!(loaded.entries.len(), OFFSET_DIRECTORY_BUFFER * 2); + } + #[compio::test] async fn given_numeric_directory_and_torn_file_when_loading_should_remove_only_invalid_file() { let dir = tempfile::tempdir().unwrap(); diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 0b104e349..6ce661ee7 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -3325,6 +3325,12 @@ mod tests { .plane .partitions() .with_partition(&namespace, |partition| { + partition.seed_recovered_consumer_offset( + iggy_common::ConsumerKind::ConsumerGroup, + 0, + 0, + 0, + ); partition.consumer_group_offsets.pin().insert( iggy_common::ConsumerGroupId(0), iggy_common::ConsumerOffset::new( @@ -3378,6 +3384,12 @@ mod tests { .plane .partitions() .with_partition(&namespace, |partition| { + partition.seed_recovered_consumer_offset( + iggy_common::ConsumerKind::ConsumerGroup, + 0, + 0, + 0, + ); partition.consumer_group_offsets.pin().insert( iggy_common::ConsumerGroupId(0), iggy_common::ConsumerOffset::new( @@ -3460,6 +3472,8 @@ mod tests { { let partitions = shard.plane.partitions(); let partition = partitions.get_by_ns(&ns).expect("partition materialised"); + partition.seed_recovered_consumer_offset(ConsumerKind::ConsumerGroup, dead_key, 7, 7); + partition.seed_recovered_consumer_offset(ConsumerKind::ConsumerGroup, live_key, 9, 9); partition.consumer_group_offsets.pin().insert( ConsumerGroupId(dead_key as usize), ConsumerOffset::new(ConsumerKind::ConsumerGroup, dead_key, 7, String::new()), diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 9ae7ec543..ddce5cd3c 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -49,7 +49,7 @@ use iggy_binary_protocol::{ #[cfg(feature = "simulator")] use iggy_common::PartitionStats; use iggy_common::variadic; -use iggy_common::{IggyError, IggyExpiry, IggyTimestamp}; +use iggy_common::{ConsumerKind, IggyError, IggyExpiry, IggyTimestamp}; use journal::superblock::{PingPongSuperblock, SuperblockStore}; use journal::{Journal, JournalHandle}; use message_bus::client_listener::RequestHandler; @@ -366,6 +366,14 @@ pub enum PartitionReadReply { stored: Option<u64>, current_offset: u64, }, + /// The read was refused and returns no messages, even where fragments were + /// already gathered. For a poll with `auto_commit`, `TooManyConsumerOffsets` + /// when the poll needed a new offset key past `[partition] + /// consumer_offsets_max`, and `TransientNotAccepted` when the auto-commit + /// could not be submitted: the owning shard's inbox was full, or the + /// partition changed primary or incarnation during the read. Transient + /// refusal permits re-polling. A capacity refusal needs a slot reclaimed + /// or a higher configured limit before a new key can succeed. Rejected(IggyError), /// Reply to [`PartitionRead::GroupOffsetState`]: the group's last-polled and /// committed offsets on this partition (each `None` if absent). @@ -2043,7 +2051,7 @@ where .ok_or(PartitionSubmitRefused)?; sender.try_send(frame).map_err(|error| { self.metrics.record_frame_drop( - crate::metrics::frame_drop_variant::PARTITION, + crate::metrics::frame_drop_variant::PARTITION_AUTO_COMMIT, crate::coordinator::classify_try_send_err(&error), ); PartitionSubmitRefused @@ -7088,6 +7096,14 @@ where continue; }; partition.retry_consumer_offset_reservations(); + if partition.queued_requests_ready() { + if walks < PARTITION_WALKS_PER_TICK_MAX { + walks += 1; + partition.resume_queued_requests().await; + } else { + walk_cursor.get_or_insert(namespace); + } + } let consensus_normal = partition.consensus().is_normal(); let consensus_view = partition.consensus().view(); let commit_min = partition.consensus().commit_min(); @@ -7459,6 +7475,21 @@ where // retires whatever completed. self.partition_repairs_inflight .set(repairs_live + repair_arms); + // Republished per sweep like the repair count: a stranded key is + // permanent until its own store or delete succeeds, so a gauge that + // never falls is the operator's only signal. + let mut stranded = [0usize; 2]; + for namespace in partitions.namespaces() { + if let Some(partition) = partitions.get_by_ns(namespace) { + stranded[0] += partition.stranded_consumer_offset_count(ConsumerKind::Consumer); + stranded[1] += + partition.stranded_consumer_offset_count(ConsumerKind::ConsumerGroup); + } + } + self.metrics + .set_consumer_offsets_stranded(ConsumerKind::Consumer, stranded[0]); + self.metrics + .set_consumer_offsets_stranded(ConsumerKind::ConsumerGroup, stranded[1]); fatal } diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 91b6a2001..c8c584436 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -39,6 +39,7 @@ use prometheus_client::encoding::EncodeLabelSet; use prometheus_client::metrics::counter::Counter; use prometheus_client::metrics::family::Family; +use prometheus_client::metrics::gauge::Gauge; use prometheus_client::registry::Registry; use std::sync::{Arc, OnceLock}; @@ -106,6 +107,12 @@ pub mod frame_drop_variant { /// dropped; the shard-0 deadline expiry recovers the slot / pending /// entry, so this stays informational. pub const REPLICA_HANDSHAKE_ACK: &str = "replica_handshake_ack"; + /// A poll's auto-commit submit refused by the owning shard's own inbox. + /// + /// Its own series, not `PARTITION`: the poll is answered with a retriable + /// status and no frame of the client's was dropped, so counting it with + /// shed frames would read as a routing loss. + pub const PARTITION_AUTO_COMMIT: &str = "partition_auto_commit"; } /// Reason labels used in `frame_drops_total`. @@ -154,7 +161,7 @@ pub mod frame_drop_reason { // pair enters the `Family` (and therefore the scrape) the first time a drop // site actually produces it, so the unreachable corners of the 7 x 9 cross // product never appear as permanent zero-valued series. -const VARIANT_COUNT: usize = 7; +const VARIANT_COUNT: usize = 8; const REASON_COUNT: usize = 11; const VARIANTS: [&str; VARIANT_COUNT] = [ @@ -165,6 +172,7 @@ const VARIANTS: [&str; VARIANT_COUNT] = [ frame_drop_variant::FORWARD_REPLICA_SEND, frame_drop_variant::METADATA_COMMIT_TICK, frame_drop_variant::REPLICA_HANDSHAKE_ACK, + frame_drop_variant::PARTITION_AUTO_COMMIT, ]; const REASONS: [&str; REASON_COUNT] = [ @@ -230,6 +238,8 @@ pub struct ShardMetrics { client_requests_denied_queue_full_total: Counter, partition_consumer_offsets_denied_total: Family<ConsumerOffsetKindLabel, Counter>, consumer_offset_denied_counters: [Counter; 2], + partition_consumer_offsets_stranded: Family<ConsumerOffsetKindLabel, Gauge>, + consumer_offset_stranded_gauges: [Gauge; 2], } impl ShardMetrics { @@ -257,6 +267,19 @@ impl ShardMetrics { .clone() }; let consumer_offset_denied_counters = [consumer_denied, consumer_group_denied]; + let partition_consumer_offsets_stranded: Family<ConsumerOffsetKindLabel, Gauge> = + Family::default(); + // End each Family read guard before creating the next series, which + // needs the same family's write lock on a miss. + let consumer_stranded = partition_consumer_offsets_stranded + .get_or_create(&ConsumerOffsetKindLabel { kind: "consumer" }) + .clone(); + let group_stranded = partition_consumer_offsets_stranded + .get_or_create(&ConsumerOffsetKindLabel { + kind: "consumer_group", + }) + .clone(); + let consumer_offset_stranded_gauges = [consumer_stranded, group_stranded]; Self { frame_drops_total, cached_counters, @@ -274,13 +297,28 @@ impl ShardMetrics { client_requests_denied_queue_full_total: Counter::default(), partition_consumer_offsets_denied_total, consumer_offset_denied_counters, + partition_consumer_offsets_stranded, + consumer_offset_stranded_gauges, } } + /// Best effort: counts explicit client denials read off the reply status + /// and the poll-side reservation refusals. A denial the pump answers to an + /// auto-commit submit has no client reply to read and is not counted. pub fn record_consumer_offset_denied(&self, kind: ConsumerKind) { self.consumer_offset_denied_counters[consumer_kind_index(kind)].inc(); } + /// Republished by every partition sweep: the sum over this shard's + /// partitions of offset keys whose file could not be loaded or unlinked. + /// Such a key stays counted against `consumer_offsets_max` until a later + /// store or delete of it succeeds, so a non-zero value that never falls is + /// an offsets directory an operator has to repair. + pub fn set_consumer_offsets_stranded(&self, kind: ConsumerKind, count: usize) { + self.consumer_offset_stranded_gauges[consumer_kind_index(kind)] + .set(i64::try_from(count).unwrap_or(i64::MAX)); + } + #[cfg(test)] #[must_use] pub fn consumer_offset_denied_value(&self, kind: ConsumerKind) -> u64 { @@ -639,9 +677,16 @@ impl ShardMetrics { ); registry.register( "partition_consumer_offsets_denied", - "consumer offset creations denied at the per-partition admission limit", + "consumer offset creations denied at the per-partition admission limit (best effort: \ + explicit client denials and poll-side reservation refusals)", self.partition_consumer_offsets_denied_total.clone(), ); + registry.register( + "partition_consumer_offsets_stranded", + "consumer offset keys whose file could not be loaded or unlinked, still counted \ + against the limit", + self.partition_consumer_offsets_stranded.clone(), + ); } } diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 242d4004f..208558d36 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1347,7 +1347,7 @@ impl Simulator { // replicated (the serving shard's job in the real server). Offset discarded. let (fragments, _commit_offset, auto_commit) = futures::executor::block_on(plan.execute())?; if let Some(applied) = auto_commit { - applied.mark_served(); + applied.admit(|_| Ok::<(), IggyError>(()))?; } Ok(fragments) }
