numinnex commented on code in PR #4063:
URL: https://github.com/apache/iggy/pull/4063#discussion_r3941072550
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2260,30 +2425,55 @@ where
|value| PendingConsumerOffsetCommit::upsert(kind, consumer_id,
value),
);
- if let Err(error) = self.persist_consumer_offset_commit(pending).await
{
- emit_partition_diag(
- tracing::Level::WARN,
- &PartitionDiagEvent::new(self.diag_ctx(), "no_ack offset
persist failed")
- .with_operation(request_header.operation)
- .with_error(error.to_string()),
- );
- return;
+ let persisted = async {
+ self.persist_consumer_offset_commit(pending).await?;
+ self.flush_consumer_offset_directories().await
}
- if let Err(error) = self.apply_consumer_offset_commit(pending) {
+ .await;
+ if let Err(error) = persisted {
+ if offset.is_some() {
+ self.release_consumer_offset_reservation(kind, consumer_id);
+ if !self.durable_consumer_offsets.contains(kind, consumer_id)
+ && let Some(path) = self.persisted_offset_path(kind,
consumer_id)
+ {
+ if delete_persisted_offset(&path).await.is_ok() {
Review Comment:
**Rollback unlink is never directory-fsynced — silent message skip on
single-replica.**
The committed-delete path now arms `consumer_offset_dirs_dirty` (`:2205`)
and flushes before replies, but this rollback unlink does not, and `:2205` is
the only arming site.
This rolls back a store the client was told **failed**
(`send_partition_deny_or_log` below). A crash after the unlink resurrects the
dirent, boot re-seeds the live map and `durable` from it, and the consumer's
stored offset becomes the value the *failed* store carried — ahead of the last
acknowledged one. The next `next()` poll resumes above it and the intervening
messages are never delivered.
Worse in kind than the delete-resurrection this fix targeted: that one
produced re-delivery (at-least-once, which the system documents), this produces
loss. Reachable only when `replica_count() == 1`, which is the topology with no
peer to heal from.
Arming the bit here (and on the create in `persist_consumer_offset_commit`)
lets the existing flush at `:2430` cover it. Same three lines also cover the
permanent `record_stranded` at `:2443` — a failed dir-fsync there burns a quota
slot with no clearing path but a later op for that same id.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2808,11 +2969,20 @@ where
.into_iter()
.chain(self.consumer_group_offsets_path.clone())
{
- if fsync_dir(&dir).await.is_err() {
+ if retry_offset_io(|| fsync_dir(&dir)).await.is_err() {
Review Comment:
**Retrying a bare fsync is not a retry — it can report success over a real
failure.**
`fsync_dir` opens the directory fresh on every call (`:1654`), and
`retry_offset_io` invokes the closure up to `OFFSET_IO_ATTEMPTS` times
discarding each error (`:1631-1636`). Linux reports a writeback/metadata error
to the first fsync that observes it and then clears it; since 4.13 the
`errseq_t` counter seeds a newly opened file's `f_wb_err` from the *current*
value at open time, so an fd opened after the error is never told about it.
Attempt 1 fails on fd A, attempt 2 opens fd B seeded post-error and returns 0.
Consequence here: a masked failure leaves `offsets_written == true`, so the
`:2977` guard never fires and the install reports success with the old offset
files' unlinks not durable — which is the resurrection shape this PR added that
guard to close. Same masking at `:3151` lets converge return `Ok` instead of
fencing.
The neighbouring `retry_offset_io(|| persist_offset(...))` at `:2941` is
correct to retry, because `persist_offset` re-opens with `truncate(true)` and
rewrites the record before its `sync_data()` — the retried fsync has freshly
dirtied pages. Retry the mutation+fsync as a unit, or treat a standalone
`fsync_dir` error as terminal.
Related: `retry_offset_io` discards the error entirely (`:1633`), so no
caller can distinguish `NotFound` from `ENOSPC` from `EIO`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4470,6 +5065,30 @@ where
self.drain_request_queue_into_prepares(drained_count).await;
}
+ async fn flush_consumer_offset_directories(&self) -> Result<(), IggyError>
{
+ for (index, dir) in [
+ self.consumer_offsets_path.as_deref(),
+ self.consumer_group_offsets_path.as_deref(),
+ ]
+ .into_iter()
+ .enumerate()
+ {
+ if !self.consumer_offset_dirs_dirty[index].get() {
+ continue;
+ }
+ if let Some(dir) = dir {
+ crate::state_transfer::fsync_dir(dir)
+ .await
+ .map_err(|_| IggyError::CannotSyncFile)?;
Review Comment:
**`map_err` discards `io::ErrorKind`, and the fatal escalates process-wide.**
Every failure here collapses to `IggyError::CannotSyncFile`, so the caller
cannot distinguish `NotFound` — nothing left to make durable — from `EIO`, a
real divergence. The error then becomes `FatalCommit`, which the shard tick
escalates to a whole-process exit, so one partition's offsets-dir open failure
takes down every partition on every shard.
Fail-closed is right for a genuine I/O error: this flush gates the delete's
reply and `advance_commit_min`, and acking a delete whose unlink is not durable
is the bug this fix closes. But the same call is WARN-only at the purge site
(`:6395`) and `.exists()`-guarded in converge (`state_transfer.rs:3150`) —
three handlings of one fault in one PR.
Propagating the kind would allow a `NotFound` carve-out and a per-partition
fence with a typed deny, instead of a process exit.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2008,54 +2118,100 @@ 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?;
- self.apply_consumer_offset_commit(pending)?;
+ self.apply_consumer_offset_commit(pending);
self.pending_consumer_offset_commits.remove(&op);
+ self.refresh_consumer_offset_reservation(pending.kind,
pending.consumer_id);
Ok(())
}
async fn persist_consumer_offset_commit(
&self,
pending: PendingConsumerOffsetCommit,
) -> Result<(), IggyError> {
- let Some(path) = self.persisted_offset_path(pending.kind,
pending.consumer_id) else {
- return Ok(());
- };
- let key = (pending.kind, pending.consumer_id);
+ let path = self.persisted_offset_path(pending.kind,
pending.consumer_id);
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ let creates_group = pending.kind == ConsumerKind::ConsumerGroup
+ && !self
+ .durable_consumer_offsets
+ .contains(pending.kind, pending.consumer_id);
match pending.mutation {
// A server auto-commit persists monotonically: its op offset can
// trail the durably-recorded value (disk-tier polls replicate in
// IO-completion order), so a plain overwrite would rewind the file
- // and re-deliver on restart. The `persisted_offsets` tracker keeps
+ // and re-deliver on restart. The durable offset tracker keeps
// the fold off the file: a covered offset skips the write, an
// advancing one blind-writes, and only a cold key (first commit
// after boot) reads the file once. Explicit client stores
// overwrite, so a deliberate offset reset still holds. Mirrors the
// in-memory `upsert_offset_max` vs `upsert_offset` split in the
// commit-apply.
PendingConsumerOffsetMutation::Upsert(offset) if
pending.auto_commit => {
- let tracked =
self.persisted_offsets.borrow().get(&key).copied();
- let persisted = match tracked {
- Some(high_water) if offset <= high_water => return Ok(()),
- Some(_) => {
- persist_offset(&path, offset,
self.consumer_offset_enforce_fsync).await?;
- offset
+ let tracked = self
+ .durable_consumer_offsets
+ .get(pending.kind, pending.consumer_id);
+ let persisted_high_water = match (path.as_deref(), tracked) {
+ (None, _) => offset,
+ (Some(_), Some(state))
+ if state
+ .persisted_high_water
+ .is_some_and(|high_water| offset <= high_water) =>
+ {
+ state.persisted_high_water.expect("covered persisted
value")
+ }
+ (Some(path), Some(state)) => {
+ let value = state.committed_offset.max(offset);
+ persist_offset(path, value,
self.consumer_offset_enforce_fsync).await?;
+ value
}
- None => {
- persist_offset_max(&path, offset,
self.consumer_offset_enforce_fsync)
- .await?
+ (Some(path), None) => {
+ persist_offset_max(path, offset,
self.consumer_offset_enforce_fsync).await?
}
};
- self.persisted_offsets.borrow_mut().insert(key, persisted);
+ self.durable_consumer_offsets.record_auto_commit(
+ pending.kind,
+ pending.consumer_id,
+ if tracked.is_none() {
+ persisted_high_water
+ } else {
+ offset
+ },
+ persisted_high_water,
+ );
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Upsert(offset) => {
- persist_offset(&path, offset,
self.consumer_offset_enforce_fsync).await?;
- self.persisted_offsets.borrow_mut().insert(key, offset);
+ if let Some(path) = path.as_deref() {
+ persist_offset(path, offset,
self.consumer_offset_enforce_fsync).await?;
+ }
+ self.durable_consumer_offsets.record_explicit(
+ pending.kind,
+ pending.consumer_id,
+ offset,
+ Some(offset),
+ );
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Delete => {
- delete_persisted_offset(&path).await?;
- self.persisted_offsets.borrow_mut().remove(&key);
+ if let Some(path) = path.as_deref() {
+ delete_persisted_offset(path).await?;
+ self.consumer_offset_dirs_dirty[match pending.kind {
Review Comment:
**Dirty bit armed even when the unlink removed nothing.**
`delete_persisted_offset` tolerates `NotFound`, so a committed delete for a
key that has no file on disk still arms the bit and buys a directory fsync in
the flush. Backup-created phantoms are exactly that shape, and the reconciler
submits them in batches of `GROUP_OFFSET_DELETES_PER_PASS`.
On its own that is wasted work. Combined with the error handling at `:5082`,
it also manufactures the precondition for a process exit on a directory that is
legitimately gone.
Arming only when a file was actually removed fixes both.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2808,11 +2969,20 @@ where
.into_iter()
.chain(self.consumer_group_offsets_path.clone())
{
- if fsync_dir(&dir).await.is_err() {
+ if retry_offset_io(|| fsync_dir(&dir)).await.is_err() {
offsets_written = false;
}
}
+ if !offsets_written {
Review Comment:
**One failed 8-byte offset write discards the entire installed partition and
re-pulls it.**
This `Err` routes to `converge_to_empty_after_failed_install`
(`:2453-2473`), which sweeps every `.log`/`.index`/staging/anchor file out of
the partition dir (`:3169-3207`), retires every log storage, and re-seeds the
counters. The original error then propagates to the catch-all arm in
`shard/src/lib.rs:8960`, which re-arms the transfer against the next peer.
So a deterministic local fault — ENOSPC, EIO — loops transfer → wipe →
transfer at the backoff rate, re-reading the serving primary's whole segment
chain every cycle. That is the amplifier `abandon_or_rearm_partition_transfer`
exists to prevent, and the offset files are 8 bytes each and independent of the
segment swap.
Writing and fsyncing the offset files *before* the segment rename would make
this failure abort pre-mutation, so the partition keeps what it had instead of
discarding a good chain.
(The converge itself is correct and thorough — `durable_consumer_offsets` is
cleared at `:3135`, so no file-less membership survives. The issue is only how
much it destroys for how small a fault.)
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2341,7 +2616,11 @@ where
}
};
- if found {
+ if found
+ || self
Review Comment:
**Replicated stranded-key deletes can crash-loop a pre-upgrade peer during a
rolling restart.**
Admitting a delete on `is_stranded` alone means no replica holds a live-map
entry for it. `dead_consumer_group_offset_ids` now extends with
`stranded_ids()`, and the reconciler submits those
(`partition_reconciler.rs:1027`), so these replicated deletes are generated
with no operator action — a torn group-offset file for a deleted group is
enough.
A pre-PR peer replaying that committed op as non-follower still runs `if
!removed && !self.consensus.is_follower() { return Err(ConsumerOffsetNotFound)
}` (`6c47193fe:iggy_partition.rs:2154`, `:2168`) → `FatalCommit` → node exit.
Why this is not narrow: there is no replica-plane version gate anywhere —
the only version negotiation is client↔server in
`binary_protocol/src/version.rs`, and grepping `core/consensus` and
`core/server` finds no peer handshake version or op-set gate. The Helm chart
defaults to RollingUpdate (`helm/charts/iggy/README.md:551`), so a
version-skewed group is the normal upgrade state, and during it the pre-upgrade
pod is the *stable* one and so a likely election winner. The reconciler
re-submits from `stranded_ids()` every pass, so the trigger is regenerated
rather than consumed until each delete commits.
Reclaiming stranded files by local unlink would avoid this entirely — a
stranded file is per-replica disk state and arguably has no business in the
replicated log. Failing that, this needs the coordinated-restart note the chart
already uses for `server.cluster.auth` (`README.md:494`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
