You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/09/07 08:39:13 UTC
(iggy) branch offset_hardening updated (265ce92a5 -> 5bc3ced7e)
This is an automated email from the ASF dual-hosted git repository.
spetz pushed a change to branch offset_hardening
in repository https://gitbox.apache.org/repos/asf/iggy.git
from 265ce92a5 Merge branch 'master' into offset_hardening
add 3462fed5c fix(shard): widen partition repair when a view-adopted suffix is missing bodies (#4074)
new 6024de1a2 Merge remote-tracking branch 'origin/master' into offset_hardening
new 5bc3ced7e hardening
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/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(-)
(iggy) 02/02: hardening
Posted by pi...@apache.org.
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 <pi...@gmail.com>
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)
}
(iggy) 01/02: Merge remote-tracking branch 'origin/master' into offset_hardening
Posted by pi...@apache.org.
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 6024de1a22554c63d07871b0b626c89377630e9d
Merge: 265ce92a5 3462fed5c
Author: spetz <pi...@gmail.com>
AuthorDate: Mon Sep 7 09:06:40 2026 +0200
Merge remote-tracking branch 'origin/master' into offset_hardening
# Conflicts:
# core/shard/src/lib.rs