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 <[email protected]> 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.
