You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/08/07 10:46:29 UTC
(iggy) 02/02: fix(server-ng): close remaining purge data-loss and typed-error gaps
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/server-ng-sdk-behavior-gaps
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit be93463d15915b9b6f38352a2813398d470e969e
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Aug 7 12:43:55 2026 +0200
fix(server-ng): close remaining purge data-loss and typed-error gaps
Several purge paths still resurrected or destroyed data. A replica
that had not applied a committed purge accepted repair floors from
purged peers and silently dropped every post-purge batch behind its
stale durable line; repaired appends bypassed the purge floor
invariant; a pre-purge truncation watermark survived the purge and
could delete post-purge segments; a purge.gen left behind by a
failed teardown made a recreated partition swallow its next purge;
and fenced journal entries stayed poll-visible and pinned memory on
idle purged partitions.
Defer repair replies until the committed purge applies locally,
guard repaired-append accounting behind the floor, zero the
watermark in both purge applies, key purge.gen to the partition's
created_revision so a stale incarnation hydrates as zero, filter
fenced entries out of resident polls, and evict the fenced prefix
up to commit_min (evicting past the walk would wedge it on a hole).
Also reject zero-count partition changes pre-consensus (legacy
create_topic admits zero, so only the change paths gate), deny
get_consumer_offset typed on a missing partition, pass
PartitionNotFound through HTTP polls, resolve the ServerDefault
size sentinel against the node default at cleaner enforcement
time, and count deferred repair serves.
---
.../integration/tests/server/poll_semantics_vsr.rs | 63 +++-
.../tests/server/topic_admission_vsr.rs | 16 +
core/metadata/src/impls/metadata.rs | 10 +
core/metadata/src/stm/stream.rs | 83 ++++-
core/partitions/src/iggy_partition.rs | 351 ++++++++++++++++++++-
core/partitions/src/journal.rs | 31 +-
core/partitions/src/offset_storage.rs | 165 ++++++++--
core/partitions/src/state_transfer.rs | 7 +-
core/server-ng/src/bootstrap.rs | 4 +
core/server-ng/src/dispatch.rs | 65 +++-
core/server-ng/src/http/handlers.rs | 11 +-
core/server-ng/src/partition_helpers.rs | 7 +-
core/server-ng/src/partition_reconciler.rs | 3 +
core/server-ng/src/responses.rs | 52 ++-
core/server-ng/src/segment_cleaner.rs | 98 +++++-
core/shard/src/lib.rs | 39 ++-
core/shard/src/metrics.rs | 19 ++
17 files changed, 951 insertions(+), 73 deletions(-)
diff --git a/core/integration/tests/server/poll_semantics_vsr.rs b/core/integration/tests/server/poll_semantics_vsr.rs
index 0078084c1..1f48eef36 100644
--- a/core/integration/tests/server/poll_semantics_vsr.rs
+++ b/core/integration/tests/server/poll_semantics_vsr.rs
@@ -17,9 +17,10 @@
//! Poll semantics against server-ng (vsr): a poll aimed at a partition id the
//! topic does not have must surface a typed `PartitionNotFound`, not an empty
-//! poll a consumer would read as end-of-partition; a timestamp poll must be
-//! at-or-after, including the message stamped exactly at the queried
-//! timestamp (the timestamp replies report per message).
+//! poll a consumer would read as end-of-partition; the same addressing error
+//! on `get_consumer_offset` must not decode as "no offset stored"; and a
+//! timestamp poll must be at-or-after, including the message stamped exactly at
+//! the queried timestamp (the timestamp replies report per message).
use iggy::prelude::*;
use integration::iggy_harness;
@@ -86,6 +87,62 @@ async fn given_missing_partition_when_polling_should_reject_partition_not_found(
assert_eq!(valid.messages.len(), 0, "empty topic polls empty");
}
+/// `get_consumer_offset` answered an unknown partition with an empty body,
+/// which the SDK decodes as `None` - the same value a consumer that simply has
+/// no stored offset yet gets back, so a client could not tell a typo from a
+/// fresh consumer. Legacy swallows this one too; server-ng surfaces the code
+/// the poll path already surfaces for the identical addressing error.
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn given_missing_partition_when_getting_consumer_offset_should_reject_partition_not_found(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ client
+ .create_stream("offset-stream")
+ .await
+ .expect("create stream");
+ let stream_id = Identifier::from_str_value("offset-stream").expect("stream identifier");
+ client
+ .create_topic(
+ &stream_id,
+ "offset-topic",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("create topic");
+ let topic_id = Identifier::from_str_value("offset-topic").expect("topic identifier");
+ let consumer = Consumer::default();
+
+ let result = client
+ .get_consumer_offset(&consumer, &stream_id, &topic_id, Some(7))
+ .await;
+
+ let expected =
+ IggyError::PartitionNotFound(7, Identifier::default(), Identifier::default()).as_code();
+ assert!(
+ matches!(&result, Err(error) if error.as_code() == expected),
+ "get_consumer_offset on partition 7 of a 1-partition topic must surface \
+ Err(PartitionNotFound), got {result:?}"
+ );
+
+ // The existing partition still answers "no offset stored" as `None`.
+ let stored = client
+ .get_consumer_offset(&consumer, &stream_id, &topic_id, Some(0))
+ .await
+ .expect("get_consumer_offset on the existing partition still succeeds");
+ assert!(
+ stored.is_none(),
+ "a consumer with no stored offset reads back as None, not an error"
+ );
+}
+
#[iggy_harness(
test_client_transport = [Tcp],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
diff --git a/core/integration/tests/server/topic_admission_vsr.rs b/core/integration/tests/server/topic_admission_vsr.rs
index ffd5b46ab..fe7bf394c 100644
--- a/core/integration/tests/server/topic_admission_vsr.rs
+++ b/core/integration/tests/server/topic_admission_vsr.rs
@@ -222,6 +222,22 @@ async fn given_out_of_bounds_partitions_count_when_mutating_should_reject_typed(
"oversized delete_partitions must deny with TooManyPartitions, got {result:?}"
);
+ // Zero is a no-op that still burns a replicated log entry, bumps the
+ // metadata revision and forces a rebalance pass. Legacy rejects it with the
+ // same code in both handlers (`1..=MAX` on create, `== 0` on delete); note
+ // that create_topic is deliberately NOT included, since a zero-partition
+ // topic is legal there in legacy too.
+ let result = client.create_partitions(&stream_id, &topic_id, 0).await;
+ assert!(
+ matches!(&result, Err(error) if error.as_code() == too_many),
+ "create_partitions with 0 must deny with TooManyPartitions, got {result:?}"
+ );
+ let result = client.delete_partitions(&stream_id, &topic_id, 0).await;
+ assert!(
+ matches!(&result, Err(error) if error.as_code() == too_many),
+ "delete_partitions with 0 must deny with TooManyPartitions, got {result:?}"
+ );
+
client
.create_partitions(&stream_id, &topic_id, 2)
.await
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index f564367ac..9643493fa 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -873,6 +873,16 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> {
self.default_max_topic_size.set(max_topic_size_bytes);
}
+ /// Byte value a stored `MaxTopicSize::ServerDefault` resolves to on this
+ /// node. Read by the per-shard segment cleaner, which enforces retention
+ /// locally and so must resolve the sentinel at enforcement time: create
+ /// admission rewrites it before replication, but an UPDATE back to
+ /// `ServerDefault` leaves the sentinel in committed state.
+ #[must_use]
+ pub const fn default_max_topic_size(&self) -> u64 {
+ self.default_max_topic_size.get()
+ }
+
/// Raise the forced-checkpoint margin to cover a configured
/// prepare-queue depth (`[metadata] prepare_queue_depth`). Clamped to
/// the built-in floor by the coordinator; no-op on shards without a
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index a59cfccbe..0a97ddc8e 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -99,7 +99,9 @@ pub struct Partition {
/// Replicated delete watermark: the reconciler on every replica removes
/// sealed segments with `end_offset` below this. Advanced monotonically by
/// `TruncatePartition` (the resolved form of a client `DeleteSegments`).
- /// `0` means nothing has been trimmed.
+ /// `0` means nothing has been trimmed. Monotone only WITHIN one offset
+ /// space: a purge restarts offsets at 0 and clears this back to 0, or the
+ /// stale watermark would keep re-staging trims over post-purge segments.
pub deleted_up_to_offset: u64,
/// Replicated purge counter: `PurgeTopic` increments it for every partition
/// in the topic. The reconciler on every replica resets a partition to a
@@ -1506,10 +1508,11 @@ impl StateHandler for PurgeStreamRequest {
type State = StreamsInner;
fn apply(&self, state: &mut StreamsInner, _timestamp: IggyTimestamp) -> ApplyReply {
// Stream purge = topic purge over every topic in the stream: advance
- // each partition's monotonic purge generation; every replica's
- // reconciler observes the committed generation and resets the
- // partition to a single empty segment at offset 0 with cleared
- // offsets (see `PurgeTopicRequest`). Metadata shape stays intact.
+ // each partition's monotonic purge generation and clear the delete
+ // watermark; every replica's reconciler observes the committed
+ // generation and resets the partition to a single empty segment at
+ // offset 0 with cleared offsets (see `PurgeTopicRequest`). Metadata
+ // shape stays intact.
let advanced = {
let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else {
return ApplyReply::err(PurgeStreamResult::StreamNotFound);
@@ -1521,6 +1524,7 @@ impl StateHandler for PurgeStreamRequest {
for (_, topic) in &mut stream.topics {
for partition in &mut topic.partitions {
partition.purge_generation = partition.purge_generation.wrapping_add(1);
+ partition.deleted_up_to_offset = 0;
advanced = true;
}
}
@@ -1762,6 +1766,12 @@ impl StateHandler for PurgeTopicRequest {
// each partition's monotonic purge generation, which the reconciler
// observes (committed generation > locally applied) and turns into a
// single empty segment at offset 0 plus cleared offsets.
+ //
+ // The delete watermark is replicated state describing the PRE-purge
+ // offset space, so it is cleared in the same apply: the purge restarts
+ // offsets at 0 and drops the consumer-offset barrier that bounded the
+ // trim, and the reconciler re-stages any nonzero watermark on every
+ // pass -- a surviving one would delete post-purge segments.
let advanced = {
let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else {
return ApplyReply::err(PurgeTopicResult::StreamNotFound);
@@ -1777,6 +1787,7 @@ impl StateHandler for PurgeTopicRequest {
};
for partition in &mut topic.partitions {
partition.purge_generation = partition.purge_generation.wrapping_add(1);
+ partition.deleted_up_to_offset = 0;
}
!topic.partitions.is_empty()
};
@@ -2468,6 +2479,68 @@ mod tests {
assert_eq!(inner.items.len(), 1);
}
+ /// A purge restarts the offset space at 0, so a watermark from the old one
+ /// must not survive: the reconciler re-stages every nonzero watermark on
+ /// each pass, and the consumer-offset barrier that bounded the trim is
+ /// cleared by the purge too, so a stale watermark deletes post-purge
+ /// segments.
+ #[test]
+ fn given_truncated_partition_when_apply_purge_should_clear_delete_watermark() {
+ let mut inner = StreamsInner::new();
+ create_stream(&mut inner, "stream");
+ let create_topic = CreateTopicWithAssignmentsRequest {
+ request: make_topic_request(0, 1, "topic"),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ };
+ let _ = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::now());
+
+ let truncate = TruncatePartitionRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ partition_id: 0,
+ up_to_offset: 500,
+ };
+ let apply = StateHandler::apply(&truncate, &mut inner, IggyTimestamp::now());
+ assert_eq!(apply.code, 0);
+ assert_eq!(
+ inner.items[0].topics[0].partitions[0].deleted_up_to_offset,
+ 500
+ );
+
+ let purge = PurgeTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ };
+ let apply = StateHandler::apply(&purge, &mut inner, IggyTimestamp::now());
+ assert_eq!(apply.code, 0);
+ assert_eq!(
+ inner.items[0].topics[0].partitions[0].deleted_up_to_offset, 0,
+ "the purge must clear the pre-purge delete watermark"
+ );
+ assert_eq!(
+ inner.items[0].topics[0].partitions[0].purge_generation, 1,
+ "the purge generation still advances"
+ );
+
+ // Same for the stream-wide purge, which walks every topic.
+ let _ = StateHandler::apply(&truncate, &mut inner, IggyTimestamp::now());
+ assert_eq!(
+ inner.items[0].topics[0].partitions[0].deleted_up_to_offset,
+ 500
+ );
+ let purge_stream = PurgeStreamRequest {
+ stream_id: WireIdentifier::numeric(0),
+ };
+ let _ = StateHandler::apply(&purge_stream, &mut inner, IggyTimestamp::now());
+ assert_eq!(
+ inner.items[0].topics[0].partitions[0].deleted_up_to_offset, 0,
+ "a stream purge clears the watermark on every partition it walks"
+ );
+ }
+
#[test]
fn given_missing_topic_when_apply_purge_topic_should_return_topic_not_found() {
let mut inner = StreamsInner::new();
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 63b40c20f..a40af1616 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -161,6 +161,13 @@ where
/// generation against this and resets only when it advances, so a redundant
/// reconcile pass never re-wipes a partition already at this generation.
pub(crate) applied_purge_generation: u64,
+ /// `Partition::created_revision` of the metadata row this partition was
+ /// built for (the reconciler's "epoch"). Keys the durable `purge.gen`
+ /// record: a delete whose on-disk cleanup failed leaves the directory
+ /// behind, and the recreated partition restarts its generations at 0, so
+ /// the dead incarnation's record must not hydrate. `0` for partitions built
+ /// without a metadata row (tests, in-memory storage).
+ pub(crate) created_revision: u64,
/// Highest consensus op assigned when the last purge ran. INVARIANT: every
/// journal-apply path must no-op entries with `op <= purge_floor_op`. The
/// purge keeps journal entries resident (consensus history for backups,
@@ -464,6 +471,7 @@ where
persisted_offsets: RefCell::new(HashMap::new()),
observed_view,
applied_purge_generation: 0,
+ created_revision: 0,
purge_floor_op: 0,
superblock: None,
superblock_lock: LocalGate::new(),
@@ -499,13 +507,22 @@ where
self.purge_floor_op
}
+ /// Record the metadata incarnation this partition was built for. Must run
+ /// BEFORE [`Self::hydrate_applied_purge_generation`], which keys the
+ /// durable record on it.
+ pub const fn set_created_revision(&mut self, created_revision: u64) {
+ self.created_revision = created_revision;
+ }
+
/// Seed [`Self::applied_purge_generation`] from the partition dir's
/// `purge.gen` file at build time (both fresh create and recovery walk
/// this). Absent file reads 0, so a partition that never purged and a
/// repair-rebuilt dir both start below any committed generation and the
/// reconciler re-applies the purge; a crash AFTER a purge's durable
/// generation write correctly skips the re-wipe, keeping messages
- /// appended since. No-op without a partition dir (in-memory storage).
+ /// appended since. A record left by a PREVIOUS incarnation of this
+ /// namespace reads 0 as well (see [`read_purge_generation`]). No-op
+ /// without a partition dir (in-memory storage).
///
/// # Errors
/// Propagates a real I/O failure reading `purge.gen`: booting with the
@@ -514,7 +531,8 @@ where
pub async fn hydrate_applied_purge_generation(&mut self) -> Result<(), IggyError> {
if let Some(dir) = self.partition_dir() {
let path = format!("{dir}/{PURGE_GENERATION_FILE}");
- self.applied_purge_generation = read_purge_generation(&path).await?;
+ self.applied_purge_generation =
+ read_purge_generation(&path, self.created_revision).await?;
}
Ok(())
}
@@ -2983,9 +3001,19 @@ where
if entry.header.operation != Operation::SendMessages {
return None;
}
- // Purge floor: a pre-purge send committing after the purge
- // reports no visible offsets ("send without confirmation", the
- // established degradation), which also keeps
+ // Purge floor: a pre-purge send committing after the purge is
+ // DELIBERATELY degraded to ZERO confirmations rather than
+ // failed. Its messages are genuinely gone (the purge deleted
+ // the segment they would have landed in) and no offset is left
+ // to report, so the reply carries the established "committed,
+ // no offsets to report" shape (`send_messages_reply_body`'s
+ // empty confirmation list, byte-identical to what a send
+ // without confirmation returns): the client sees success with
+ // an empty confirmations list and re-sends if it needs the
+ // offset. A typed transient status was the alternative and is
+ // wrong here -- the op DID commit cluster-wide, so telling the
+ // client to retry duplicates a committed send into the
+ // post-purge offset space. `None` is also what keeps
// `commit_partition_entry` from re-advancing the reset offset
// and stats with pre-purge values.
if entry.header.op <= self.purge_floor_op {
@@ -3989,6 +4017,23 @@ where
.journal()
.inner
.clear_poll_index(self.purge_floor_op);
+ // Hand the already-walked fenced prefix to the normal eviction path so
+ // an idle purged partition does not pin it resident: the flush that
+ // would otherwise evict it is gated on `journal.info.messages_count`,
+ // which the reset above just zeroed, so with no post-purge traffic the
+ // entries never leave. Repair semantics are unchanged -- `evict_prefix`
+ // moves them into the evicted ring, still op-addressable by
+ // `repair_entry`, and the serve path clamps `retained_from` above the
+ // floor anyway. Bounded at `commit_min`, NOT `commit_max`: an op the
+ // commit walk has not reached yet still needs its header resident, or
+ // `committed_headers_from` stops at the hole and wedges `commit_min`.
+ let fenced_prefix = self
+ .log
+ .journal()
+ .inner
+ .committed_prefix(self.consensus.commit_min().min(self.purge_floor_op))
+ .len();
+ self.evict_committed_prefix(fenced_prefix).await;
// Last durable step: record the applied generation before the
// in-memory marker advances. On a write failure the marker stays old,
@@ -4001,7 +4046,9 @@ where
// recorded the generation keep it.
if let Some(dir) = self.partition_dir() {
let path = format!("{dir}/{PURGE_GENERATION_FILE}");
- if let Err(error) = persist_purge_generation(&path, generation).await {
+ if let Err(error) =
+ persist_purge_generation(&path, generation, self.created_revision).await
+ {
self.purge_deferred = true;
warn!(
target: "iggy.partitions.diag",
@@ -4269,6 +4316,7 @@ where
let write_lock = self.write_lock.clone();
let _guard = write_lock.lock().await;
+ let op = message.header().op;
let (base_offset, base_timestamp, total_size, message_count) = {
let batch =
decode_prepare_slice(message.as_slice()).map_err(|_| IggyError::InvalidCommand)?;
@@ -4282,6 +4330,24 @@ where
if message_count == 0 {
return Ok(None);
}
+
+ // Purge floor: the same fence every other journal-apply path honors. A
+ // repaired pre-purge batch is still journaled -- the commit walk stops
+ // at the first missing op, so dropping it would wedge `commit_min` --
+ // but it must not re-advance the reset counters or re-count purged
+ // bytes. `None` also keeps it out of the session's
+ // `first_batch_offset`: that anchors the floor-connect check, and
+ // purged bytes cannot stand in for durable state.
+ if op <= self.purge_floor_op {
+ self.log
+ .journal()
+ .inner
+ .append(message.into_frozen())
+ .await
+ .map_err(|_| IggyError::CannotAppendMessage)?;
+ return Ok(None);
+ }
+
let last_offset = base_offset + u64::from(message_count) - 1;
self.should_increment_offset = true;
@@ -6654,7 +6720,7 @@ mod purge_floor_tests {
the (fresh, empty) segments"
);
assert_eq!(
- partition.log.journal().inner.resident_entries().len(),
+ partition.log.journal().inner.resident_count(),
2,
"journal entries are consensus history and must survive the purge"
);
@@ -6663,6 +6729,11 @@ mod purge_floor_tests {
&& partition.log.journal().inner.header_by_op(2).is_some(),
"repair and retransmission must still resolve pre-purge ops"
);
+ assert!(
+ partition.log.journal().inner.resident_entries().is_empty(),
+ "the poll view of the resident tier must exclude fenced entries, \
+ even though they stay resident for consensus"
+ );
let _ = std::fs::remove_dir_all(&dir);
}
@@ -6810,6 +6881,272 @@ mod purge_floor_tests {
let _ = std::fs::remove_dir_all(&dir);
}
+ /// The resident poll tier is sealed by the purge, but the first post-purge
+ /// indexed append re-arms it. The snapshot handed to the straddle and
+ /// retention-recovery walks matches on batch CONTENTS alone (no op), so
+ /// without the fence those walks serve purged bytes again.
+ #[compio::test]
+ async fn given_post_purge_append_when_snapshotting_resident_tail_should_skip_fenced_entries() {
+ let (mut partition, dir) = purge_test_partition("resident-fence");
+ // Two pre-purge batches: offsets 0 and 1 (the counter advances).
+ journal_send_batch(&mut partition, 1).await;
+ journal_send_batch(&mut partition, 2).await;
+
+ partition
+ .purge(&repair_config(), 1)
+ .await
+ .expect("purge partition");
+
+ // Re-arms the resident tier: this batch restarts at offset 0.
+ journal_send_batch(&mut partition, 3).await;
+
+ let snapshot = partition.resident_tail_snapshot();
+ assert_eq!(
+ snapshot.entries.len(),
+ 1,
+ "only the post-purge entry may reach a poll"
+ );
+ // Offset 1 existed ONLY in the purged batch, so a resident poll there
+ // must come up empty instead of serving the fenced entry.
+ let purged_offset = crate::journal::select_resident(
+ &snapshot.entries,
+ MessageLookup::Offset {
+ offset: 1,
+ count: 10,
+ ceiling: u64::MAX,
+ },
+ );
+ assert!(
+ purged_offset.is_none(),
+ "a purged offset must not be servable from the resident tier"
+ );
+ assert!(
+ crate::journal::select_resident(
+ &snapshot.entries,
+ MessageLookup::Offset {
+ offset: 0,
+ count: 10,
+ ceiling: u64::MAX,
+ },
+ )
+ .is_some(),
+ "the post-purge batch is still servable"
+ );
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
+ /// The flush that would evict the fenced prefix is gated on
+ /// `journal.info.messages_count`, which the purge zeroes, so an idle purged
+ /// partition would pin those entries resident forever. The purge hands them
+ /// to the ordinary eviction path instead; repair still resolves them from
+ /// the evicted ring.
+ #[compio::test]
+ async fn given_walked_prefix_when_purged_should_evict_fenced_entries_to_the_ring() {
+ let (mut partition, dir) = purge_test_partition("fenced-evict");
+ // Single-replica test partitions disable repair retention; the ring is
+ // what makes eviction safe for repair, so exercise it.
+ partition.log.journal().inner.set_repair_retention(true);
+ journal_send_batch(&mut partition, 1).await;
+ journal_send_batch(&mut partition, 2).await;
+ partition.consensus().advance_commit_max(2);
+ // Walked already (a settled repair floor does this without flushing),
+ // so the entries are committed history that is still resident.
+ partition.consensus().set_commit_floor(2);
+
+ partition
+ .purge(&repair_config(), 1)
+ .await
+ .expect("purge partition");
+
+ assert_eq!(
+ partition.log.journal().inner.resident_count(),
+ 0,
+ "the walked fenced prefix must not stay pinned in resident storage"
+ );
+ assert!(
+ partition.log.journal().inner.repair_entry(1).is_some()
+ && partition.log.journal().inner.repair_entry(2).is_some(),
+ "eviction moves them to the ring, where repair still resolves them"
+ );
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
+ /// `purge_floor_op` promises EVERY journal-apply path no-ops at or below the
+ /// floor. The repaired-prepare path writes the dirty offset, the segment
+ /// write cursor and `journal.info`, so it needs the same guard: only the
+ /// peer-side serve clamp kept purged bytes out, and that clamp is the
+ /// PEER's floor, not this replica's.
+ #[compio::test]
+ async fn given_repaired_send_at_or_below_floor_when_appended_should_not_mutate_state() {
+ let (mut partition, dir) = purge_test_partition("repaired-fenced");
+ journal_send_batch(&mut partition, 1).await;
+ partition
+ .purge(&repair_config(), 1)
+ .await
+ .expect("purge partition");
+ assert_eq!(partition.purge_floor_op(), 1, "the floor fences op 1");
+
+ // A repaired pre-purge batch for the fenced op, carrying its original
+ // (pre-purge) stamps exactly as the serving peer stored them.
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let record = build_segment_record(namespace, 40);
+ let header_size = std::mem::size_of::<PrepareHeader>();
+ let total = header_size + record.len();
+ let mut message = Message::<PrepareHeader>::new(total);
+ message.as_mut_slice()[header_size..].copy_from_slice(&record);
+ let message = message.transmute_header(|_, header: &mut PrepareHeader| {
+ header.command = Command2::Prepare;
+ header.operation = Operation::SendMessages;
+ header.op = 1;
+ header.namespace = namespace.inner();
+ header.size = u32::try_from(total).expect("prepare size fits u32");
+ });
+
+ let base_offset = partition
+ .append_repaired_send_messages(message)
+ .await
+ .expect("a fenced repaired prepare is journaled, not refused");
+
+ assert_eq!(
+ base_offset, None,
+ "a purged batch must not anchor the repair floor's connect check"
+ );
+ assert_eq!(
+ partition.dirty_offset.load(Ordering::Relaxed),
+ 0,
+ "the reset counter must not jump to a purged offset"
+ );
+ let segment_index = partition.log.segments().len() - 1;
+ assert_eq!(
+ partition.log.segments()[segment_index].current_position,
+ 0,
+ "no purged bytes may be reserved in the fresh segment"
+ );
+ assert_eq!(
+ partition.log.journal().info.messages_count,
+ 0,
+ "purged bytes must not re-enter the flush accounting"
+ );
+ assert!(
+ partition.log.journal().inner.header_by_op(1).is_some(),
+ "the entry is still journaled: dropping it would wedge commit_min"
+ );
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
+ /// Why the shard defers repair COMPLETION while a committed purge is
+ /// unapplied: until the purge lands, `recovered_durable_offset` still names
+ /// the pre-purge segments, and every repaired post-purge batch (offsets
+ /// restart at 0) silently vanishes in the flush skip. The purge clears the
+ /// line, and the same batch persists.
+ #[compio::test]
+ async fn given_stale_recovered_durable_offset_when_committing_should_drop_until_purge_applies()
+ {
+ let (mut partition, dir) = purge_test_partition("stale-durable");
+ // A restart that recovered segments through offset 9.
+ partition.recovered_durable_offset = Some(9);
+
+ journal_send_batch(&mut partition, 1).await;
+ partition.consensus().advance_commit_max(1);
+ partition.commit_journal(&repair_config()).await;
+ assert_eq!(
+ partition.log.active_segment().size.as_bytes_u64(),
+ 0,
+ "a batch at offset 0 is skipped as already-durable while the stale \
+ recovered line stands"
+ );
+
+ partition
+ .purge(&repair_config(), 1)
+ .await
+ .expect("purge partition");
+ assert_eq!(
+ partition.recovered_durable_offset, None,
+ "the purge deleted those bytes, so the line must go with them"
+ );
+
+ journal_send_batch(&mut partition, 2).await;
+ partition.consensus().advance_commit_max(2);
+ partition.commit_journal(&repair_config()).await;
+ assert_eq!(
+ partition.log.active_segment().size.as_bytes_u64(),
+ build_segment_record(IggyNamespace::new(1, 1, 0), 0).len() as u64,
+ "after the purge the same offset-0 batch reaches the segment"
+ );
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
+ /// A delete whose on-disk cleanup failed leaves the partition directory
+ /// (and `purge.gen`) behind. The recreated topic's rows restart their purge
+ /// generations at 0, so hydrating the DEAD incarnation's generation would
+ /// swallow the new topic's purges until the committed counter climbed past
+ /// it.
+ #[compio::test]
+ async fn given_purge_gen_from_a_dead_incarnation_when_rebuilt_should_hydrate_zero() {
+ let (mut partition, dir) = purge_test_partition("stale-incarnation");
+ partition.set_created_revision(7);
+ partition
+ .purge(&repair_config(), 4)
+ .await
+ .expect("purge partition");
+ assert_eq!(partition.applied_purge_generation(), 4);
+
+ let rebuild = |created_revision: u64| {
+ let mut rebuilt = test_partition();
+ rebuilt.set_partition_dir(dir.to_string_lossy().into_owned());
+ rebuilt.set_created_revision(created_revision);
+ rebuilt
+ };
+
+ // Same incarnation (an ordinary restart): the generation still stands.
+ let mut restarted = rebuild(7);
+ restarted
+ .hydrate_applied_purge_generation()
+ .await
+ .expect("hydrate purge generation");
+ assert_eq!(restarted.applied_purge_generation(), 4);
+
+ // New incarnation over the same directory.
+ let mut recreated = rebuild(8);
+ recreated
+ .hydrate_applied_purge_generation()
+ .await
+ .expect("hydrate purge generation");
+ assert_eq!(
+ recreated.applied_purge_generation(),
+ 0,
+ "a dead incarnation's record must not fence the recreated partition"
+ );
+
+ // So the new topic's first purge (generation 1) passes the reconciler's
+ // `committed > applied` gate and re-keys the record.
+ recreated
+ .purge(&repair_config(), 1)
+ .await
+ .expect("purge partition");
+ let mut after = rebuild(8);
+ after
+ .hydrate_applied_purge_generation()
+ .await
+ .expect("hydrate purge generation");
+ assert_eq!(after.applied_purge_generation(), 1);
+ let mut dead = rebuild(7);
+ dead.hydrate_applied_purge_generation()
+ .await
+ .expect("hydrate purge generation");
+ assert_eq!(
+ dead.applied_purge_generation(),
+ 0,
+ "re-keying leaves the dead incarnation with nothing to hydrate"
+ );
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
#[compio::test]
async fn purge_persists_generation_and_hydrates_it_back() {
let (mut partition, dir) = purge_test_partition("generation");
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index f48f71a1d..f29a951f2 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -646,13 +646,34 @@ impl PartitionJournal<PartitionJournalMemStorage> {
inner.storage.is_empty()
}
- /// Owned, op-ascending clones of every resident journal entry. Each clone
- /// is a `Frozen` refcount bump, not a deep copy. Used to snapshot the
- /// resident tail at poll-plan time so a disk-tier straddle can be spliced
- /// off the partition borrow on owned data ([`crate::iggy_partition`]).
+ /// Owned, op-ascending clones of the resident journal entries a poll may
+ /// serve. Each clone is a `Frozen` refcount bump, not a deep copy. Used to
+ /// snapshot the resident tail at poll-plan time so a disk-tier straddle can
+ /// be spliced off the partition borrow on owned data
+ /// ([`crate::iggy_partition`]).
+ ///
+ /// Entries at or below the purge floor are filtered out. They stay resident
+ /// (consensus history for backups, repair and retransmission) but are
+ /// poll-fenced exactly like the offset/timestamp indexes
+ /// [`Self::clear_poll_index`] sealed: the snapshot walk matches on the batch
+ /// contents alone, so an unfiltered list re-exposes purged bytes as soon as
+ /// one post-purge append puts an entry back into the index.
pub fn resident_entries(&self) -> Vec<JournalBuffer> {
let inner = unsafe { &*self.inner.get() };
- inner.storage.entries()
+ let entries = inner.storage.entries();
+ let floor = self.poll_floor.get();
+ if floor == 0 {
+ return entries;
+ }
+ // `headers[i]` pairs with storage index `i` (see the length-lock
+ // invariant on `append_with_meta`), so the op comes from the header
+ // vector rather than a per-entry decode.
+ let headers = unsafe { &*self.headers.get() };
+ headers
+ .iter()
+ .zip(entries)
+ .filter_map(|(header, entry)| (header.op > floor).then_some(entry))
+ .collect()
}
}
diff --git a/core/partitions/src/offset_storage.rs b/core/partitions/src/offset_storage.rs
index 1e357c988..c3377d824 100644
--- a/core/partitions/src/offset_storage.rs
+++ b/core/partitions/src/offset_storage.rs
@@ -21,13 +21,19 @@ use compio::{
};
use iggy_common::IggyError;
use std::path::Path;
+use tracing::warn;
const OFFSET_SIZE: usize = core::mem::size_of::<u64>();
/// Per-partition file recording the purge generation this replica last applied
-/// locally (LE u64, in the partition dir beside the segments it fences).
+/// locally, in the partition dir beside the segments it fences. Two LE u64s:
+/// the applied generation, then the `created_revision` of the partition
+/// incarnation it was applied for.
pub const PURGE_GENERATION_FILE: &str = "purge.gen";
+/// `[generation][created_revision]`, both LE u64.
+const PURGE_GENERATION_RECORD_SIZE: usize = 2 * OFFSET_SIZE;
+
pub async fn persist_offset(path: &str, offset: u64, enforce_fsync: bool) -> Result<(), IggyError> {
// No `exists()` probe first: that is a BLOCKING `std::path` stat on the pump
// in front of every write, which serialises a batched fan-out on stats
@@ -87,33 +93,107 @@ pub async fn persist_offset_max(
Ok(effective)
}
-/// Durably record the purge generation a partition has locally applied. Same
-/// layout as [`persist_offset`] (LE u64, truncate+write) but ALWAYS data-synced,
-/// regardless of the consumer-offset fsync knob: purges are rare, the file is
-/// 8 bytes, and a generation lost from the page cache in a crash makes the
-/// reconciler re-purge on restart, wiping messages appended after the purge.
-/// A failure leaves the previous generation on disk so the caller keeps its
-/// in-memory applied generation old and retries.
+/// Durably record the purge generation a partition has locally applied, keyed
+/// to the incarnation (`created_revision`) it was applied for. Truncate+write
+/// like [`persist_offset`] but ALWAYS data-synced, regardless of the
+/// consumer-offset fsync knob: purges are rare, the record is 16 bytes, and a
+/// generation lost from the page cache in a crash makes the reconciler
+/// re-purge on restart, wiping messages appended after the purge. A failure
+/// leaves the previous record on disk so the caller keeps its in-memory
+/// applied generation old and retries.
///
/// # Errors
/// Propagates the underlying open/write/sync failure.
-pub async fn persist_purge_generation(path: &str, generation: u64) -> Result<(), IggyError> {
- persist_offset(path, generation, true).await
+pub async fn persist_purge_generation(
+ path: &str,
+ generation: u64,
+ created_revision: u64,
+) -> Result<(), IggyError> {
+ if let Some(parent) = Path::new(path).parent() {
+ create_dir_all(parent).await.map_err(|_| {
+ IggyError::CannotCreateConsumerOffsetsDirectory(parent.display().to_string())
+ })?;
+ }
+
+ let mut file = OpenOptions::new()
+ .write(true)
+ .create(true)
+ .truncate(true)
+ .open(path)
+ .await
+ .map_err(|_| IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
+ let mut record = [0u8; PURGE_GENERATION_RECORD_SIZE];
+ record[..OFFSET_SIZE].copy_from_slice(&generation.to_le_bytes());
+ record[OFFSET_SIZE..].copy_from_slice(&created_revision.to_le_bytes());
+ file.write_all_at(record, 0)
+ .await
+ .0
+ .map_err(|_| IggyError::CannotWriteToFile)?;
+ file.sync_data()
+ .await
+ .map_err(|_| IggyError::CannotWriteToFile)?;
+ Ok(())
}
-/// Read the persisted purge generation. Absent and torn files map to `Ok(0)`:
-/// both imply a purge died mid-write, and `0` makes the reconciler re-apply
-/// the purge, the correct self-healing recovery for an idempotent wipe. A
-/// real I/O error propagates instead: collapsing it to `0` would re-purge a
+/// Read the purge generation this replica applied for the `created_revision`
+/// incarnation of the partition.
+///
+/// Absent and torn files map to `Ok(0)`: both imply a purge died mid-write, and
+/// `0` makes the reconciler re-apply the purge, the correct self-healing
+/// recovery for an idempotent wipe.
+///
+/// A record written for a DIFFERENT incarnation maps to `Ok(0)` too. A failed
+/// `delete_partitions_from_disk` leaves the directory (and this file) behind;
+/// the recreated topic's generations restart at 0, so hydrating the dead
+/// incarnation's generation would swallow every purge of the new topic until
+/// the committed counter climbed past it.
+///
+/// A real I/O error propagates instead: collapsing it to `0` would re-purge a
/// partition whose durable generation is intact but momentarily unreadable,
/// destroying every message appended after that purge.
///
/// # Errors
/// Propagates a real open/read failure (anything but absent or short).
-pub async fn read_purge_generation(path: &str) -> Result<u64, IggyError> {
- read_persisted_offset(path)
+pub async fn read_purge_generation(path: &str, created_revision: u64) -> Result<u64, IggyError> {
+ if !Path::new(path).exists() {
+ return Ok(0);
+ }
+ let file = OpenOptions::new()
+ .read(true)
+ .open(path)
.await
- .map(|offset| offset.unwrap_or(0))
+ .map_err(|_| IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
+ let buf = vec![0u8; PURGE_GENERATION_RECORD_SIZE];
+ let compio::BufResult(read, buf) = file.read_exact_at(buf, 0).await;
+ match read {
+ Ok(()) => {}
+ Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(0),
+ Err(_) => return Err(IggyError::CannotReadConsumerOffsets(path.to_owned())),
+ }
+ let (generation_bytes, revision_bytes) = buf.split_at(OFFSET_SIZE);
+ let generation = u64::from_le_bytes(
+ generation_bytes
+ .try_into()
+ .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?,
+ );
+ let stored_revision = u64::from_le_bytes(
+ revision_bytes
+ .try_into()
+ .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?,
+ );
+ if stored_revision != created_revision {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ path,
+ generation,
+ stored_revision,
+ created_revision,
+ "ignoring a purge generation recorded for another partition incarnation"
+ );
+ return Ok(0);
+ }
+ Ok(generation)
}
/// Read a single persisted consumer offset. `None` if the file is absent or
@@ -239,23 +319,23 @@ mod tests {
.into_owned();
assert_eq!(
- read_purge_generation(&path).await.expect("absent file"),
+ read_purge_generation(&path, 11).await.expect("absent file"),
0,
"absent file is 0"
);
- persist_purge_generation(&path, 3)
+ persist_purge_generation(&path, 3, 11)
.await
.expect("persist generation");
assert_eq!(
- read_purge_generation(&path).await.expect("valid file"),
+ read_purge_generation(&path, 11).await.expect("valid file"),
3,
"round-trip"
);
std::fs::write(&path, [0xAB, 0xCD]).expect("write torn file");
assert_eq!(
- read_purge_generation(&path).await.expect("torn file"),
+ read_purge_generation(&path, 11).await.expect("torn file"),
0,
"torn file degrades to 0 so the reconciler re-applies the purge"
);
@@ -263,7 +343,7 @@ mod tests {
// A directory path is a real I/O error, not a short read: it must
// surface, not collapse to the re-purge sentinel (a silent re-purge
// would destroy post-purge messages).
- let result = read_purge_generation(&dir.to_string_lossy()).await;
+ let result = read_purge_generation(&dir.to_string_lossy(), 11).await;
assert!(
matches!(result, Err(IggyError::CannotReadConsumerOffsets(_))),
"real I/O error must propagate, got {result:?}",
@@ -272,6 +352,47 @@ mod tests {
let _ = std::fs::remove_dir_all(&dir);
}
+ /// A failed `delete_partitions_from_disk` leaves the directory and this
+ /// file behind. The recreated partition's generations restart at 0, so a
+ /// record from the DEAD incarnation must not be hydrated: it would swallow
+ /// every purge of the new topic until the committed counter climbed past
+ /// it.
+ #[compio::test]
+ async fn purge_generation_from_another_incarnation_reads_as_zero() {
+ let dir = unique_temp_dir();
+ let path = dir
+ .join(PURGE_GENERATION_FILE)
+ .to_string_lossy()
+ .into_owned();
+
+ persist_purge_generation(&path, 9, 41)
+ .await
+ .expect("persist generation");
+
+ assert_eq!(
+ read_purge_generation(&path, 41).await.expect("same dir"),
+ 9,
+ "the incarnation that wrote it still hydrates it"
+ );
+ assert_eq!(
+ read_purge_generation(&path, 42).await.expect("stale file"),
+ 0,
+ "a record from a dead incarnation must not fence the new one"
+ );
+
+ // The new incarnation's own purge re-keys the file.
+ persist_purge_generation(&path, 1, 42)
+ .await
+ .expect("persist generation");
+ assert_eq!(read_purge_generation(&path, 42).await.expect("rekeyed"), 1);
+ assert_eq!(
+ read_purge_generation(&path, 41).await.expect("now stale"),
+ 0
+ );
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
#[compio::test]
async fn persist_offset_max_recovers_torn_file() {
let dir = unique_temp_dir();
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 54ba925f3..22bb8cd86 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -2566,7 +2566,12 @@ where
&& let Some(dir) = self.partition_dir.clone()
{
let path = format!("{dir}/{PURGE_GENERATION_FILE}");
- if let Err(error) = persist_purge_generation(&path, offsets_wire.purge_generation).await
+ if let Err(error) = persist_purge_generation(
+ &path,
+ offsets_wire.purge_generation,
+ self.created_revision,
+ )
+ .await
{
tracing::warn!(
target: "iggy.partitions.diag",
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 6c2f55d0f..ad1671b20 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -1801,6 +1801,7 @@ async fn build_shard_for_thread(
config,
namespace,
partition_stats,
+ partition_metadata.created_revision,
topology.cluster_id,
topology.self_replica_id,
topology.replica_count,
@@ -2359,6 +2360,9 @@ async fn load_partition(
config.partition.evicted_ring_bytes_max.as_bytes_u64(),
);
partition.set_partition_dir(partition_dir);
+ // Before the hydrate: the durable record is keyed by incarnation, so a
+ // `purge.gen` left behind by a previous life of this namespace reads 0.
+ partition.set_created_revision(partition_metadata.created_revision);
partition.hydrate_applied_purge_generation().await?;
hydrate_partition_log(
&mut partition,
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 5769d0e99..7e64bf7de 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -740,11 +740,15 @@ fn pop_next_client_request(
message
}
-/// Per-request partitions-count cap shared by create-topic, create-partitions
+/// Per-request partitions-count cap, shared by create-topic, create-partitions
/// and delete-partitions admission. Runs pre-consensus like
/// [`validate_topic_bounds`]: an oversized count must not burn a replicated
/// log entry (create-partitions admission would also allocate that many
/// consensus-group ids before replicating).
+///
+/// Zero passes here because a zero-partition TOPIC is legal (legacy
+/// `create_topic` admits `0..=MAX`); the add/remove requests reject it in
+/// [`validate_partitions_change_count`].
pub(crate) const fn validate_partitions_count(partitions_count: u32) -> Result<(), IggyError> {
if partitions_count > MAX_PARTITIONS_PER_REQUEST {
return Err(IggyError::TooManyPartitions);
@@ -752,6 +756,21 @@ pub(crate) const fn validate_partitions_count(partitions_count: u32) -> Result<(
Ok(())
}
+/// [`validate_partitions_count`] plus the zero rejection that create-partitions
+/// and delete-partitions carry: adding or removing zero partitions is a no-op
+/// that would still burn a replicated log entry, bump `Streams::revision` and
+/// force every shard through a rebalance pass. Legacy rejects it with
+/// `TooManyPartitions` in both handlers (`1..=MAX` on create, `== 0` on
+/// delete), so the code matches rather than inventing a new one.
+pub(crate) const fn validate_partitions_change_count(
+ partitions_count: u32,
+) -> Result<(), IggyError> {
+ if partitions_count == 0 {
+ return Err(IggyError::TooManyPartitions);
+ }
+ validate_partitions_count(partitions_count)
+}
+
/// Static create-topic bounds shared by the TCP and HTTP ingresses. Runs
/// pre-consensus: a rejected request must not burn a replicated log entry,
/// and `prepare_request` errors evict the session instead of denying typed.
@@ -1044,12 +1063,12 @@ async fn handle_client_request<B, MJ, S, SB>(
Operation::CreatePartitions => CreatePartitionsRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
.and_then(|create_partitions| {
- validate_partitions_count(create_partitions.partitions_count)
+ validate_partitions_change_count(create_partitions.partitions_count)
}),
Operation::DeletePartitions => DeletePartitionsRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
.and_then(|delete_partitions| {
- validate_partitions_count(delete_partitions.partitions_count)
+ validate_partitions_change_count(delete_partitions.partitions_count)
}),
_ => Ok(()),
};
@@ -1992,6 +2011,20 @@ async fn handle_get_consumer_offset<B, MJ, S, SB>(
_ => Bytes::new(),
}
}
+ // A partition id that does not exist in a resolvable topic is a client
+ // addressing error, the same one the poll path denies typed. An empty
+ // body decodes as `None` -- indistinguishable from "this consumer has
+ // no stored offset yet" -- so the caller cannot tell a typo from a
+ // fresh consumer.
+ Err(error @ IggyError::PartitionNotFound(..)) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "get_consumer_offset rejected: partition not found"
+ );
+ send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await;
+ return;
+ }
Err(error) => {
warn!(
transport_client_id,
@@ -3683,5 +3716,31 @@ mod tests {
),
"one past the cap must deny"
);
+ // Zero passes the shared cap because a zero-partition TOPIC is legal
+ // (legacy `create_topic` admits `0..=MAX`).
+ assert!(validate_partitions_count(0).is_ok());
+ }
+
+ #[test]
+ fn zero_partitions_change_denies_pre_consensus() {
+ // Adding or removing zero partitions is a no-op that would still burn
+ // a replicated log entry and force a rebalance. Legacy rejects it with
+ // `TooManyPartitions` in both handlers, so the code matches.
+ assert!(
+ matches!(
+ validate_partitions_change_count(0),
+ Err(IggyError::TooManyPartitions)
+ ),
+ "adding or removing zero partitions must deny"
+ );
+ assert!(validate_partitions_change_count(1).is_ok());
+ assert!(validate_partitions_change_count(MAX_PARTITIONS_PER_REQUEST).is_ok());
+ assert!(
+ matches!(
+ validate_partitions_change_count(MAX_PARTITIONS_PER_REQUEST + 1),
+ Err(IggyError::TooManyPartitions)
+ ),
+ "the cap still applies"
+ );
}
}
diff --git a/core/server-ng/src/http/handlers.rs b/core/server-ng/src/http/handlers.rs
index 4513dc715..0dc160fe2 100644
--- a/core/server-ng/src/http/handlers.rs
+++ b/core/server-ng/src/http/handlers.rs
@@ -1035,9 +1035,16 @@ pub(in crate::http) async fn poll_messages(
Err(IggyError::ConsumerGroupPartitionNotOwned(..)) => {
return Ok(Json(resync_required_polled_messages()));
}
+ // A partition id the topic does not have is a client addressing
+ // error with its own code; collapsing it into the generic 404 body
+ // told an SDK "no such stream/topic" for a request whose stream and
+ // topic both resolved. TCP parity: the dispatch denies typed here.
+ Err(error @ IggyError::PartitionNotFound(..)) => {
+ return Err(ReadError::Rejected(error));
+ }
// The remaining resolver failures are STM lookups that came up
- // empty (unknown stream, topic, partition, or consumer group), so
- // they render as the legacy 404 body.
+ // empty (unknown stream, topic, or consumer group), so they render
+ // as the legacy 404 body.
Err(_) => return Err(ReadError::NotFound),
};
let reply = SendWrapper::new(
diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs
index 9edb1e571..a7a291eb7 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -575,10 +575,12 @@ pub(crate) fn restore_partition_view(
///
/// Returns [`ServerNgError`] when bounds validation, directory creation,
/// superblock recovery, or segment provisioning fails.
+#[allow(clippy::too_many_arguments)]
pub async fn build_partition_fresh(
config: &ServerNgConfig,
namespace: IggyNamespace,
stats: Arc<PartitionStats>,
+ created_revision: u64,
cluster_id: u128,
self_replica_id: u8,
replica_count: u8,
@@ -683,7 +685,10 @@ pub async fn build_partition_fresh(
// Fresh dirs read generation 0; a dir surviving from a crashed process
// (this "fresh" build races repair re-materialization) reads the last
// durably-applied purge so the reconciler does not re-wipe messages
- // appended after it.
+ // appended after it. Keyed by incarnation, so a dir left behind by a failed
+ // delete does not fence the recreated partition's purges: set the revision
+ // first.
+ partition.set_created_revision(created_revision);
partition.hydrate_applied_purge_generation().await?;
partition.created_at = IggyTimestamp::now();
partition.offset.store(0, Ordering::Release);
diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs
index 2bca5b9dd..63d6c2a5c 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -507,6 +507,8 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
stale = counters.stale,
deferred = counters.deferred,
parked_reclaimed = counters.parked_reclaimed,
+ purges_staged = counters.purges_staged,
+ trims_pending = counters.trims_pending,
"partition reconciler pass complete"
);
} else {
@@ -630,6 +632,7 @@ async fn reconcile_additions(
ctx.config.as_ref(),
ns,
partition_stats,
+ epoch,
ctx.cluster_id,
ctx.self_replica_id,
ctx.replica_count,
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index e9d98d40c..7463e81ea 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -1192,14 +1192,21 @@ fn partition_response(
partition: &metadata::stm::stream::Partition,
) -> Result<PartitionResponse, IggyError> {
// Per-partition counters live in the shared stats registry (one `Arc`
- // across all shards and both left-right buffers), populated when the
- // owning shard materializes the partition; `None` only in the window
- // before that first materialization.
+ // across all shards and both left-right buffers).
//
- // A committed partition always materializes with exactly one empty
- // segment, so before the owning shard gets there (registry miss, or
- // registered but not yet segmented) the reply reports that deterministic
- // initial state instead of a zero a client would read as "no storage".
+ // Registration is NOT materialization: the owning shard's reconciler mints
+ // the entry (get-or-create in `fetch_partition_stats`) before it builds the
+ // partition, and `ensure_initial_segment` only bumps `segments_count` once
+ // the segment file is open. So a registry MISS and a registered entry still
+ // reading zero segments are the same thing to a caller -- committed, not yet
+ // holding storage -- and both report the deterministic shape every
+ // materialization lands on: one empty segment at offset 0. A bare zero
+ // would read as "no storage" to a client polling right after `create_topic`.
+ //
+ // Cost of the clamp: a partition fenced for rebuild (tombstoned after a
+ // refused chain) also reads as one empty segment rather than zero. Telling
+ // the two apart needs a materialization signal the registry does not carry
+ // today; the counters are still the honest source for size and messages.
let stats = streams
.stats_registry
.partition_get(stream_id, topic_id, partition.id);
@@ -1728,6 +1735,37 @@ mod tests {
assert!(!stats.hostname.is_empty());
}
+ #[test]
+ fn partition_response_reports_the_initial_shape_until_a_segment_exists() {
+ use iggy_common::{StreamStats, TopicStats};
+ use metadata::stm::stream::{Partition, StreamsInner};
+
+ let streams = StreamsInner::new();
+ let partition = Partition::new(0, 1, IggyTimestamp::from(1u64), 0);
+
+ // Registry miss: the owning shard has not started building.
+ let predicted = partition_response(&streams, 0, 0, &partition).expect("response builds");
+ assert_eq!(predicted.segments_count, 1);
+ assert_eq!(predicted.messages_count, 0);
+
+ // Registered but not yet segmented: the reconciler mints the entry
+ // before `ensure_initial_segment` runs, so this is the SAME state to a
+ // caller and must not read as "no storage".
+ let topic_stats = Arc::new(TopicStats::new(Arc::new(StreamStats::default())));
+ let stats = streams.stats_registry.partition(0, 0, 0, topic_stats);
+ let mid_build = partition_response(&streams, 0, 0, &partition).expect("response builds");
+ assert_eq!(mid_build.segments_count, 1);
+
+ // Materialized: the real counters answer from here on.
+ stats.increment_segments_count(1);
+ stats.increment_messages_count(7);
+ stats.increment_size_bytes(64);
+ let live = partition_response(&streams, 0, 0, &partition).expect("response builds");
+ assert_eq!(live.segments_count, 1);
+ assert_eq!(live.messages_count, 7);
+ assert_eq!(live.size_bytes, 64);
+ }
+
#[test]
fn topic_header_echoes_stored_size_and_expiry_verbatim() {
use iggy_common::{
diff --git a/core/server-ng/src/segment_cleaner.rs b/core/server-ng/src/segment_cleaner.rs
index 9e0e104a7..bffd01616 100644
--- a/core/server-ng/src/segment_cleaner.rs
+++ b/core/server-ng/src/segment_cleaner.rs
@@ -61,6 +61,9 @@ fn stage_owned_partitions(shard: &Rc<ServerNgShard>) {
let now = IggyTimestamp::now();
let namespaces: Vec<_> = shard.plane.partitions().namespaces().copied().collect();
let streams = shard.plane.metadata().mux_stm.streams();
+ // Resolved once per pass, not per partition: the node default is a `Cell`
+ // written at bootstrap and never after.
+ let default_max_topic_size = shard.plane.metadata().default_max_topic_size();
for namespace in namespaces {
let Some((message_expiry, max_topic_size, partition_count)) =
streams.topic_retention_config(namespace.stream_id(), namespace.topic_id())
@@ -75,18 +78,8 @@ fn stage_owned_partitions(shard: &Rc<ServerNgShard>) {
message_expiry,
IggyExpiry::NeverExpire | IggyExpiry::ServerDefault
);
- let max_bytes = match max_topic_size {
- // Per-partition budget: the cluster has no single owner of a
- // topic-wide total, so each partition keeps an equal share.
- MaxTopicSize::Custom(size) => {
- let divisor = u64::try_from(partition_count).unwrap_or(1).max(1);
- Some(size.as_bytes_u64() / divisor)
- }
- // No per-partition cap. `ServerDefault` must NOT fall through to a
- // sized branch: its `as_bytes_u64()` is 0, which would trim every
- // sealed segment. The server-ng default topic size is unlimited.
- MaxTopicSize::Unlimited | MaxTopicSize::ServerDefault => None,
- };
+ let max_bytes =
+ per_partition_size_budget(max_topic_size, default_max_topic_size, partition_count);
if !has_expiry && max_bytes.is_none() {
continue;
@@ -94,3 +87,84 @@ fn stage_owned_partitions(shard: &Rc<ServerNgShard>) {
shard.request_clean_partition(namespace, now, message_expiry, max_bytes);
}
}
+
+/// Per-partition byte budget for a topic, or `None` for "no cap".
+///
+/// The cluster has no single owner of a topic-wide total, so each partition
+/// keeps an equal share.
+///
+/// `ServerDefault` is resolved against the node default HERE, at enforcement
+/// time. Create admission rewrites the sentinel before replication, but an
+/// UPDATE to `ServerDefault` leaves it in committed state, and reading that as
+/// "no cap" made an updated topic behave differently from an identically
+/// configured created one. A node default of unlimited (the shipped config)
+/// still yields `None`.
+///
+/// `ServerDefault` must never reach a sized branch: its `as_bytes_u64()` is 0,
+/// which would trim every sealed segment.
+fn per_partition_size_budget(
+ max_topic_size: MaxTopicSize,
+ default_max_topic_size: u64,
+ partition_count: usize,
+) -> Option<u64> {
+ let resolved = match max_topic_size {
+ MaxTopicSize::ServerDefault => MaxTopicSize::from(default_max_topic_size),
+ sized => sized,
+ };
+ match resolved {
+ MaxTopicSize::Custom(size) => {
+ let divisor = u64::try_from(partition_count).unwrap_or(1).max(1);
+ Some(size.as_bytes_u64() / divisor)
+ }
+ // `From<u64>` maps 0 back to `ServerDefault`, so a node default of 0
+ // lands here as "no cap" rather than as a trim-everything budget.
+ MaxTopicSize::Unlimited | MaxTopicSize::ServerDefault => None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::per_partition_size_budget;
+ use iggy_common::MaxTopicSize;
+
+ #[test]
+ fn server_default_resolves_to_the_node_default_at_enforcement_time() {
+ // A topic UPDATED back to `ServerDefault` keeps the sentinel in
+ // committed state; the cleaner must enforce the node default anyway, or
+ // it diverges from a topic CREATED with the same setting (whose
+ // sentinel admission already rewrote).
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::ServerDefault, 4096, 4),
+ Some(1024),
+ "the node default is resolved and split across partitions"
+ );
+ // Shipped config: server default is unlimited -> still no cap.
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::ServerDefault, u64::MAX, 4),
+ None
+ );
+ // A zero node default must read as "no cap", never as a zero budget
+ // that trims every sealed segment.
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::ServerDefault, 0, 4),
+ None
+ );
+ }
+
+ #[test]
+ fn explicit_sizes_ignore_the_node_default() {
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::Custom(4096u64.into()), 64, 2),
+ Some(2048)
+ );
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::Unlimited, 64, 2),
+ None
+ );
+ // Zero partitions must not divide by zero.
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::Custom(4096u64.into()), 64, 0),
+ Some(4096)
+ );
+ }
+}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index f9bc7cf87..54d1fb246 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3829,6 +3829,7 @@ where
namespace.partition_id(),
);
if committed_purge > partition.applied_purge_generation() {
+ self.metrics.record_partition_repair_serve_deferred();
tracing::debug!(
shard = self.id,
namespace_raw = header.namespace,
@@ -4149,11 +4150,8 @@ where
0
};
let config = planes.1.0.config().clone();
- let Some(partition) = planes
- .1
- .0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
- else {
+ let namespace = IggyNamespace::from_raw(header.namespace);
+ let Some(partition) = planes.1.0.get_mut_by_ns(&namespace) else {
return;
};
let Some(session) = partition.repair else {
@@ -4162,6 +4160,37 @@ where
if header.nonce != session.nonce {
return;
}
+ // Receiver half of the serve-side purge gate: while a committed purge
+ // is not yet locally applied, this replica's `recovered_durable_offset`
+ // still describes the PRE-purge segments, so a floor from a peer that
+ // did purge reads as connected against state the purge is about to
+ // delete -- and the post-purge batches (offsets restarting at 0) then
+ // flush-skip below that stale durable line and are silently lost.
+ // Defer the whole reply: the purge is one reconciler wake away and
+ // resets the line to `None`, and the stall retry re-asks, so the peer
+ // re-emits both `RangeEvicted` and `RepairDone` for the same window.
+ let committed_purge = self
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .partition_purge_generation(
+ namespace.stream_id(),
+ namespace.topic_id(),
+ namespace.partition_id(),
+ );
+ if committed_purge > partition.applied_purge_generation() {
+ self.metrics.record_partition_repair_serve_deferred();
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ committed_purge,
+ applied_purge = partition.applied_purge_generation(),
+ command = ?header.command,
+ "deferring repair completion until the committed purge applies locally"
+ );
+ return;
+ }
match header.command {
Command2::RangeEvicted => {
if let Some(repair) = partition.repair.as_mut() {
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index 883eefe2d..659550217 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -190,6 +190,7 @@ pub struct ShardMetrics {
partition_frames_rejected_stale_total: Counter,
partition_frames_rejected_ahead_total: Counter,
partition_requests_denied_transient_total: Counter,
+ partition_repair_serves_deferred_purge_total: Counter,
}
impl ShardMetrics {
@@ -223,6 +224,7 @@ impl ShardMetrics {
partition_frames_rejected_stale_total: Counter::default(),
partition_frames_rejected_ahead_total: Counter::default(),
partition_requests_denied_transient_total: Counter::default(),
+ partition_repair_serves_deferred_purge_total: Counter::default(),
}
}
@@ -366,6 +368,23 @@ impl ShardMetrics {
self.partition_requests_denied_transient_total.get()
}
+ /// Bumped every time this replica declines to serve or complete a partition
+ /// repair because a committed purge has not applied locally yet. One or two
+ /// per rejoin is the normal convergence window; a sustained climb means the
+ /// purge never landed, and the requester is spinning its stall retry with
+ /// nothing but a `debug!` to show for it.
+ pub fn record_partition_repair_serve_deferred(&self) {
+ self.partition_repair_serves_deferred_purge_total.inc();
+ }
+
+ /// Snapshot of `partition_repair_serves_deferred_purge_total`.
+ /// Test/simulator accessor.
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn partition_repair_serves_deferred_purge_value(&self) -> u64 {
+ self.partition_repair_serves_deferred_purge_total.get()
+ }
+
/// Snapshot of `partition_frames_rejected_stale_total`. Test/simulator
/// accessor, readable from any crate under those cfgs so the crates that
/// drive the reconciler can assert a reject did not happen.