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 4f5b088cf398f38761ba569751eca7f32bc02176 Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Aug 7 10:12:52 2026 +0200 fix(server-ng): close server-ng SDK-visible behavior gaps The Python SDK suite pinned twelve server-ng divergences from the legacy server as strict xfails; this closes the eleven that were server bugs (get_topics on a missing stream erroring is intended). Timestamp polls now filter on the broker append time replies report; a missing partition id denies with PartitionNotFound instead of an empty success; an unmaterialized partition reports its initial one-empty-segment state. Pre-consensus validation shared by TCP and HTTP denies TooManyPartitions (create_topic, create_partitions, delete_partitions) and InvalidTopicSize before a rejected request burns a replicated log entry. Updates store ServerDefault sentinels verbatim and reads echo the stored size and expiry instead of freezing the node default into state. Purge never reached partition data. Purge commits now wake the reconciler; the applied generation is durable per partition (purge.gen, always fsynced last, and an I/O error reading it back fails boot instead of silently re-purging); a purge floor fences every journal-apply path, surviving the poll-index rebuild of eviction re-appends; repair serves only above the floor and defers while a committed purge is not yet locally applied. Repair-vs-purge ordering across a StartView stays open, documented as TODO. # Conflicts: # core/partitions/src/iggy_partition.rs # core/server-ng/src/bootstrap.rs # core/server-ng/src/dispatch.rs # core/server-ng/src/partition_helpers.rs # core/server-ng/src/partition_reconciler.rs # core/server-ng/src/responses.rs # core/shard/src/lib.rs --- core/integration/tests/server/mod.rs | 11 + .../integration/tests/server/poll_semantics_vsr.rs | 189 +++++++++ core/integration/tests/server/purge_vsr.rs | 51 +++ .../server/scenarios/purge_delete_scenario.rs | 153 ++++++- .../tests/server/scenarios/system_scenario.rs | 21 +- .../tests/server/topic_admission_vsr.rs | 238 +++++++++++ core/metadata/src/impls/metadata.rs | 57 +-- core/partitions/src/iggy_partition.rs | 459 ++++++++++++++++++++- core/partitions/src/journal.rs | 129 +++++- core/partitions/src/offset_storage.rs | 75 ++++ core/partitions/src/state_transfer.rs | 46 ++- core/server-ng/src/bootstrap.rs | 38 +- core/server-ng/src/dispatch.rs | 250 ++++++++--- core/server-ng/src/http/handlers.rs | 10 +- core/server-ng/src/partition_helpers.rs | 5 + core/server-ng/src/partition_reconciler.rs | 89 +++- core/server-ng/src/responses.rs | 155 +++---- core/shard/src/lib.rs | 48 ++- core/shard/src/router.rs | 18 + 19 files changed, 1809 insertions(+), 233 deletions(-) diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs index b93b0052e..ed6ce0743 100644 --- a/core/integration/tests/server/mod.rs +++ b/core/integration/tests/server/mod.rs @@ -26,6 +26,17 @@ mod flush_vsr; // they must evict typed (MalformedLogin), not stall or reply empty-ok. #[cfg(feature = "vsr")] mod legacy_login_vsr; +// Poll addressing + timestamp semantics: typed PartitionNotFound on a bad +// partition id, at-or-after timestamp polls. +#[cfg(feature = "vsr")] +mod poll_semantics_vsr; +// Create-topic static bounds deny typed before consensus. +#[cfg(feature = "vsr")] +mod topic_admission_vsr; +// Purge durability: applied generation survives restart; journal-resident +// purged batches stay fenced behind the purge floor. +#[cfg(feature = "vsr")] +mod purge_vsr; // Shared HTTP transport plumbing (session + verb helpers) for the raw-HTTP // server-ng suites below. #[cfg(feature = "vsr")] diff --git a/core/integration/tests/server/poll_semantics_vsr.rs b/core/integration/tests/server/poll_semantics_vsr.rs new file mode 100644 index 000000000..0078084c1 --- /dev/null +++ b/core/integration/tests/server/poll_semantics_vsr.rs @@ -0,0 +1,189 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! 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). + +use iggy::prelude::*; +use integration::iggy_harness; +use tokio::time::{Duration, sleep}; + +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_missing_partition_when_polling_should_reject_partition_not_found( + harness: &TestHarness, +) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("poll-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("poll-stream").expect("stream identifier"); + client + .create_topic( + &stream_id, + "poll-topic", + 1, + CompressionAlgorithm::None, + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .expect("create topic"); + let topic_id = Identifier::from_str_value("poll-topic").expect("topic identifier"); + + let result = client + .poll_messages( + &stream_id, + &topic_id, + Some(7), + &Consumer::default(), + &PollingStrategy::offset(0), + 1, + false, + ) + .await; + + let expected = + IggyError::PartitionNotFound(7, Identifier::default(), Identifier::default()).as_code(); + assert!( + matches!(&result, Err(error) if error.as_code() == expected), + "polling partition 7 of a 1-partition topic must surface Err(PartitionNotFound), got {result:?}" + ); + + let valid = client + .poll_messages( + &stream_id, + &topic_id, + Some(0), + &Consumer::default(), + &PollingStrategy::offset(0), + 1, + false, + ) + .await + .expect("poll on the existing partition still succeeds"); + assert_eq!(valid.messages.len(), 0, "empty topic polls empty"); +} + +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_message_at_polled_timestamp_when_polling_should_include_it(harness: &TestHarness) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("ts-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("ts-stream").expect("stream identifier"); + client + .create_topic( + &stream_id, + "ts-topic", + 1, + CompressionAlgorithm::None, + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .expect("create topic"); + let topic_id = Identifier::from_str_value("ts-topic").expect("topic identifier"); + + // Two sends spaced apart so the broker stamps distinct batch timestamps. + let mut first = vec![ + IggyMessage::builder() + .payload("first".into()) + .build() + .expect("message"), + ]; + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut first, + ) + .await + .expect("send first"); + sleep(Duration::from_millis(20)).await; + let mut second = vec![ + IggyMessage::builder() + .payload("second".into()) + .build() + .expect("message"), + ]; + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut second, + ) + .await + .expect("send second"); + + let all = client + .poll_messages( + &stream_id, + &topic_id, + Some(0), + &Consumer::default(), + &PollingStrategy::offset(0), + 10, + false, + ) + .await + .expect("poll all"); + assert_eq!(all.messages.len(), 2, "both messages are readable"); + let second_timestamp = all.messages[1].header.timestamp; + assert!( + second_timestamp > all.messages[0].header.timestamp, + "spaced sends must carry distinct broker timestamps" + ); + + // Poll at the exact reported timestamp of the second message: at-or-after + // semantics must return it, not skip past it. + let polled = client + .poll_messages( + &stream_id, + &topic_id, + Some(0), + &Consumer::default(), + &PollingStrategy::timestamp(second_timestamp.into()), + 10, + false, + ) + .await + .expect("poll by timestamp"); + assert_eq!( + polled.messages.len(), + 1, + "timestamp poll at the second message's own timestamp returns exactly it" + ); + assert_eq!( + polled.messages[0].header.offset, all.messages[1].header.offset, + "the message at the queried timestamp is the one returned" + ); +} diff --git a/core/integration/tests/server/purge_vsr.rs b/core/integration/tests/server/purge_vsr.rs new file mode 100644 index 000000000..5cd802854 --- /dev/null +++ b/core/integration/tests/server/purge_vsr.rs @@ -0,0 +1,51 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! server-ng purge durability: the applied purge generation survives a +//! restart (`purge.gen`), and purged journal-resident batches stay fenced +//! behind the purge floor instead of resurfacing through the shutdown flush. + +use crate::server::scenarios::purge_delete_scenario; +use integration::iggy_harness; + +// Single node: the tests reason about ONE replica's on-disk state across a +// restart (`purge.gen` hydration, shutdown-flush fencing); a cluster would +// route polls to whichever node leads after the restart. Default fsync +// config on purpose: the restart is graceful, so segment bytes survive +// without fsync, and `purge.gen` is unconditionally synced by the purge. +#[iggy_harness( + cluster_nodes = 1, + server(partition.messages_required_to_save = "1") +)] +async fn given_post_purge_messages_when_server_restarts_should_retain_them( + harness: &mut TestHarness, +) { + purge_delete_scenario::run_purge_survives_restart(harness).await; +} + +// The huge threshold keeps every batch journal-resident: nothing is ever +// flushed before the purge, so the purged bytes exist ONLY as consensus +// history the shutdown flush re-walks. +#[iggy_harness( + cluster_nodes = 1, + server(partition.messages_required_to_save = "10000") +)] +async fn given_journal_resident_messages_when_purged_should_not_resurface( + harness: &mut TestHarness, +) { + purge_delete_scenario::run_resident_purge_no_resurface(harness).await; +} diff --git a/core/integration/tests/server/scenarios/purge_delete_scenario.rs b/core/integration/tests/server/scenarios/purge_delete_scenario.rs index 8c65c3b10..c790aae63 100644 --- a/core/integration/tests/server/scenarios/purge_delete_scenario.rs +++ b/core/integration/tests/server/scenarios/purge_delete_scenario.rs @@ -994,8 +994,9 @@ pub async fn run_purge_topic(harness: &mut TestHarness, restart_server: bool) { // landed mid-purge. There boot plants the [0] layout itself (fencing a torn // chain, or recovering an already-drained directory) with the offset files // still present, so the layout gate above is satisfied BEFORE the - // reconciler's re-purge clears them (the applied generation is not - // persisted, so a restart re-purges). Everywhere else the pump clears + // reconciler's re-purge clears them (the kill preceded the purge.gen + // record, so boot hydrates the old generation and the reconciler + // re-purges). Everywhere else the pump clears // offsets and files in the SAME frame that plants the layout, and a poll // would hide a regression that clears them one frame late. Kept short -- // a client-visible stale offset after purge-then-restart is a real @@ -1083,6 +1084,154 @@ pub async fn run_purge_topic(harness: &mut TestHarness, restart_server: bool) { client.delete_stream(&stream_ident).await.unwrap(); } +/// Messages appended AFTER a purge must survive a restart: the purge's +/// applied generation is durable (`purge.gen`), so boot re-hydrates it and +/// the reconciler does not re-apply the (still-committed) purge over the +/// post-purge data. Without that file a restart re-reads applied=0 against +/// the replayed committed generation and silently wipes the new messages on +/// its first pass. +#[cfg(feature = "vsr")] +pub async fn run_purge_survives_restart(harness: &mut TestHarness) { + let client = build_root_client(harness); + client.connect().await.unwrap(); + let data_path = harness.server().data_path().to_path_buf(); + + let stream = client.create_stream(STREAM_NAME).await.unwrap(); + let stream_ident = Identifier::named(STREAM_NAME).unwrap(); + let topic = client + .create_topic( + &stream_ident, + TOPIC_NAME, + 1, + CompressionAlgorithm::None, + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .unwrap(); + let topic_ident = Identifier::named(TOPIC_NAME).unwrap(); + let partition_path = partition_path(&data_path, stream.id, topic.id); + + send_messages(&client, &stream_ident, &topic_ident, 10).await; + client + .purge_topic(&stream_ident, &topic_ident) + .await + .unwrap(); + await_segment_layout(&partition_path, &[0]).await; + + send_messages(&client, &stream_ident, &topic_ident, 3).await; + poll_exactly(&client, &stream_ident, &topic_ident, 3).await; + + maybe_restart(harness, true).await; + + // Ride out the boot reconcile pass: an un-hydrated applied generation + // would re-purge asynchronously, so an immediate poll could still see + // the messages a moment before they vanish. + tokio::time::sleep(std::time::Duration::from_millis(1500)).await; + let polled = poll_exactly(&client, &stream_ident, &topic_ident, 3).await; + let offsets: Vec<u64> = polled.messages.iter().map(|m| m.header.offset).collect(); + assert_eq!( + offsets, + vec![0, 1, 2], + "post-purge messages must survive the restart at their offsets" + ); +} + +/// Journal-resident messages must not resurface after a purge: with the +/// flush threshold too high to ever persist, the purged batches stay in the +/// in-memory journal as consensus history, and the graceful-shutdown flush +/// walks them again. The purge floor must fence them out of the segment so +/// the restart recovers only the post-purge appends. +#[cfg(feature = "vsr")] +pub async fn run_resident_purge_no_resurface(harness: &mut TestHarness) { + let client = build_root_client(harness); + client.connect().await.unwrap(); + + client.create_stream(STREAM_NAME).await.unwrap(); + let stream_ident = Identifier::named(STREAM_NAME).unwrap(); + client + .create_topic( + &stream_ident, + TOPIC_NAME, + 1, + CompressionAlgorithm::None, + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .unwrap(); + let topic_ident = Identifier::named(TOPIC_NAME).unwrap(); + + send_messages(&client, &stream_ident, &topic_ident, 5).await; + poll_exactly(&client, &stream_ident, &topic_ident, 5).await; + + client + .purge_topic(&stream_ident, &topic_ident) + .await + .unwrap(); + // The purge is asynchronous and the segments are empty both before and + // after it (nothing ever flushed), so the poll going empty IS the + // convergence signal: it proves the resident poll tier was sealed. + poll_exactly(&client, &stream_ident, &topic_ident, 0).await; + + send_messages(&client, &stream_ident, &topic_ident, 3).await; + let polled = poll_exactly(&client, &stream_ident, &topic_ident, 3).await; + let offsets: Vec<u64> = polled.messages.iter().map(|m| m.header.offset).collect(); + assert_eq!(offsets, vec![0, 1, 2], "post-purge appends restart at 0"); + + // Graceful restart: shutdown force-flushes the committed journal, whose + // front still holds the five fenced pre-purge batches. + maybe_restart(harness, true).await; + + let polled = poll_exactly(&client, &stream_ident, &topic_ident, 3).await; + let offsets: Vec<u64> = polled.messages.iter().map(|m| m.header.offset).collect(); + assert_eq!( + offsets, + vec![0, 1, 2], + "purged resident batches must not resurface through the shutdown \ + flush or recovery" + ); +} + +/// Poll from offset 0 with headroom (count 100) until exactly `expected` +/// messages are served, so an extra resurfaced message fails the count +/// instead of being cropped by the poll size. Panics after +/// [`POLL_CONVERGENCE_TIMEOUT`] with the last observed count. +#[cfg(feature = "vsr")] +async fn poll_exactly( + client: &IggyClient, + stream_ident: &Identifier, + topic_ident: &Identifier, + expected: usize, +) -> PolledMessages { + let deadline = std::time::Instant::now() + POLL_CONVERGENCE_TIMEOUT; + loop { + let polled = client + .poll_messages( + stream_ident, + topic_ident, + Some(PARTITION_ID), + &Consumer::default(), + &PollingStrategy::offset(0), + 100, + false, + ) + .await + .unwrap(); + if polled.messages.len() == expected { + return polled; + } + assert!( + std::time::Instant::now() < deadline, + "poll did not converge to {expected} messages, last saw {}", + polled.messages.len() + ); + tokio::time::sleep(POLL_RETRY_INTERVAL).await; + } +} + /// Wait until the server-visible stored offset for `consumer` reaches /// `expected`. Auto-commit stores are issued by a detached SDK task and /// applied on the partition's owning shard, so the only ordering guarantee diff --git a/core/integration/tests/server/scenarios/system_scenario.rs b/core/integration/tests/server/scenarios/system_scenario.rs index c296ad631..2e11ba499 100644 --- a/core/integration/tests/server/scenarios/system_scenario.rs +++ b/core/integration/tests/server/scenarios/system_scenario.rs @@ -152,15 +152,18 @@ pub async fn run(harness: &TestHarness) { assert_eq!(topic.max_topic_size, MaxTopicSize::Unlimited); assert_eq!(topic.replication_factor, 1); - // 11. Get topic details by ID; the owning shards materialize the fresh - // partitions (first segment included) asynchronously after the commit. - let topic = get_topic_when(&client, STREAM_NAME, TOPIC_NAME, |topic| { - topic - .partitions - .iter() - .all(|partition| partition.segments_count == 1) - }) - .await; + // 11. Get topic details by ID. The owning shards materialize fresh + // partitions asynchronously after the commit, but the reply reports the + // deterministic initial state (one empty segment) for a committed + // partition immediately, so no convergence loop is needed here. + let topic = client + .get_topic( + &Identifier::named(STREAM_NAME).unwrap(), + &Identifier::named(TOPIC_NAME).unwrap(), + ) + .await + .unwrap() + .expect("Failed to get topic"); assert_eq!(topic.id, topic_id); assert_eq!(topic.name, TOPIC_NAME); assert_eq!(topic.partitions_count, PARTITIONS_COUNT); diff --git a/core/integration/tests/server/topic_admission_vsr.rs b/core/integration/tests/server/topic_admission_vsr.rs new file mode 100644 index 000000000..ffd5b46ab --- /dev/null +++ b/core/integration/tests/server/topic_admission_vsr.rs @@ -0,0 +1,238 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Topic admission and echo semantics against server-ng (vsr). Create bounds +//! must be rejected with typed errors before consensus: partitions count above +//! `MAX_PARTITIONS_PER_REQUEST` denies with `TooManyPartitions` (for create +//! topic, create partitions and delete partitions alike); a custom +//! `max_topic_size` below the configured segment size denies with +//! `InvalidTopicSize`; `ServerDefault` and `Unlimited` sizes pass. Update +//! stores `max_topic_size` and `message_expiry` verbatim and gets echo the +//! stored value (never the node default frozen at update time), matching +//! legacy wire behavior. + +use std::str::FromStr; + +use iggy::prelude::*; +use integration::iggy_harness; + +const PARTITIONS_LIMIT: u32 = 1000; + +async fn create_topic_with( + client: &IggyClient, + stream_id: &Identifier, + name: &str, + partitions_count: u32, + max_topic_size: MaxTopicSize, +) -> Result<TopicDetails, IggyError> { + client + .create_topic( + stream_id, + name, + partitions_count, + CompressionAlgorithm::None, + None, + IggyExpiry::NeverExpire, + max_topic_size, + ) + .await +} + +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_out_of_bounds_topic_when_creating_should_reject_typed(harness: &TestHarness) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("admission-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("admission-stream").expect("stream identifier"); + + let too_many = IggyError::TooManyPartitions.as_code(); + for partitions_count in [PARTITIONS_LIMIT + 1, 10_000] { + let result = create_topic_with( + &client, + &stream_id, + "too-many-partitions", + partitions_count, + MaxTopicSize::ServerDefault, + ) + .await; + assert!( + matches!(&result, Err(error) if error.as_code() == too_many), + "{partitions_count} partitions must deny with TooManyPartitions, got {result:?}" + ); + } + + // Below the default segment size (1 GiB) => rejected before consensus. + let tiny = MaxTopicSize::Custom(IggyByteSize::from_str("10KiB").expect("byte size")); + let result = create_topic_with(&client, &stream_id, "tiny-topic", 1, tiny).await; + let invalid_size = IggyError::InvalidTopicSize(tiny, IggyByteSize::default()).as_code(); + assert!( + matches!(&result, Err(error) if error.as_code() == invalid_size), + "max_topic_size below segment size must deny with InvalidTopicSize, got {result:?}" + ); + + create_topic_with( + &client, + &stream_id, + "boundary-partitions", + PARTITIONS_LIMIT, + MaxTopicSize::ServerDefault, + ) + .await + .expect("exactly MAX_PARTITIONS_PER_REQUEST partitions is accepted"); + + create_topic_with( + &client, + &stream_id, + "unlimited-topic", + 1, + MaxTopicSize::Unlimited, + ) + .await + .expect("unlimited max_topic_size is accepted"); +} + +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_updated_topic_when_getting_topic_should_echo_stored_values(harness: &TestHarness) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("echo-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("echo-stream").expect("stream identifier"); + create_topic_with( + &client, + &stream_id, + "echo-topic", + 1, + MaxTopicSize::Custom(IggyByteSize::from_str("2GiB").expect("byte size")), + ) + .await + .expect("create topic"); + let topic_id = Identifier::from_str_value("echo-topic").expect("topic identifier"); + + let update_topic = |max_topic_size: MaxTopicSize, message_expiry: IggyExpiry| { + let client = &client; + let stream_id = &stream_id; + let topic_id = &topic_id; + async move { + client + .update_topic( + stream_id, + topic_id, + "echo-topic", + CompressionAlgorithm::None, + None, + message_expiry, + max_topic_size, + ) + .await + .expect("update topic"); + let topic = client + .get_topic(stream_id, topic_id) + .await + .expect("get topic") + .expect("topic exists"); + (topic.max_topic_size, topic.message_expiry) + } + }; + + // The server echoes both stored sentinels as wire 0 (legacy parity). The + // SDK decodes a topic-response size 0 as `ServerDefault` but an expiry 0 + // as `NeverExpire` (`wire_conversions`), so that is the legacy-identical + // client-visible read-back; the node default must NOT leak into either. + assert_eq!( + update_topic(MaxTopicSize::ServerDefault, IggyExpiry::ServerDefault).await, + (MaxTopicSize::ServerDefault, IggyExpiry::NeverExpire), + "an update to ServerDefault must echo the stored sentinel, \ + not the node default frozen at update time" + ); + let custom_size = MaxTopicSize::Custom(IggyByteSize::from_str("3GiB").expect("byte size")); + let custom_expiry = IggyExpiry::ExpireDuration(IggyDuration::from_str("5s").expect("duration")); + assert_eq!( + update_topic(custom_size, custom_expiry).await, + (custom_size, custom_expiry), + "explicit custom values echo verbatim" + ); + assert_eq!( + update_topic(MaxTopicSize::Unlimited, IggyExpiry::NeverExpire).await, + (MaxTopicSize::Unlimited, IggyExpiry::NeverExpire), + "unlimited size and never-expire echo verbatim" + ); +} + +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_out_of_bounds_partitions_count_when_mutating_should_reject_typed( + harness: &TestHarness, +) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("partitions-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("partitions-stream").expect("stream identifier"); + create_topic_with( + &client, + &stream_id, + "partitions-topic", + 1, + MaxTopicSize::ServerDefault, + ) + .await + .expect("create topic"); + let topic_id = Identifier::from_str_value("partitions-topic").expect("topic identifier"); + + let too_many = IggyError::TooManyPartitions.as_code(); + let result = client + .create_partitions(&stream_id, &topic_id, PARTITIONS_LIMIT + 1) + .await; + assert!( + matches!(&result, Err(error) if error.as_code() == too_many), + "oversized create_partitions must deny with TooManyPartitions, got {result:?}" + ); + let result = client + .delete_partitions(&stream_id, &topic_id, PARTITIONS_LIMIT + 1) + .await; + assert!( + matches!(&result, Err(error) if error.as_code() == too_many), + "oversized delete_partitions must deny with TooManyPartitions, got {result:?}" + ); + + client + .create_partitions(&stream_id, &topic_id, 2) + .await + .expect("in-bounds create_partitions is accepted"); + let topic = client + .get_topic(&stream_id, &topic_id) + .await + .expect("get topic") + .expect("topic exists"); + assert_eq!( + topic.partitions_count, 3, + "the in-bounds add lands after the oversized denies" + ); +} diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 55d4fb042..f564367ac 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -41,7 +41,6 @@ use iggy_binary_protocol::requests::partitions::CreatePartitionsRequest as WireC use iggy_binary_protocol::requests::partitions::CreatePartitionsWithAssignmentsRequest as PersistedCreatePartitionsRequest; use iggy_binary_protocol::requests::topics::CreateTopicRequest as WireCreateTopicRequest; use iggy_binary_protocol::requests::topics::CreateTopicWithAssignmentsRequest as PersistedCreateTopicRequest; -use iggy_binary_protocol::requests::topics::UpdateTopicRequest as WireUpdateTopicRequest; use iggy_binary_protocol::{ Command2, ConsensusHeader, EvictionReason, GenericHeader, Operation, PrepareHeader, PrepareOkHeader, ReplyHeader, RequestHeader, WireDecode, WireEncode, WireName, @@ -892,25 +891,13 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> { self.client_table.borrow_mut().set_capacity(max_clients); } - /// Resolved byte value for `MaxTopicSize::ServerDefault`. - #[must_use] - pub const fn default_max_topic_size(&self) -> u64 { - self.default_max_topic_size.get() - } - /// Install the resolved micros value used for `IggyExpiry::ServerDefault`. - /// Server-ng bootstrap calls this with `system.topic.message_expiry` on every - /// shard (responses read it too); only shard 0's copy feeds admission. + /// Server-ng bootstrap calls this with `system.topic.message_expiry`; only + /// shard 0's copy feeds admission. pub fn set_default_message_expiry(&self, message_expiry_micros: u64) { self.default_message_expiry.set(message_expiry_micros); } - /// Resolved micros value for `IggyExpiry::ServerDefault`. - #[must_use] - pub const fn default_message_expiry(&self) -> u64 { - self.default_message_expiry.get() - } - /// Fire post-commit notifier. Clones the `Rc` out under a short /// borrow so a re-entrant `set_commit_notifier` from inside the /// closure cannot panic on `borrow_mut`. @@ -3321,30 +3308,10 @@ where &body, )) } - Operation::UpdateTopic => { - let mut request = WireUpdateTopicRequest::decode_from(body) - .map_err(|_| IggyError::InvalidCommand)?; - // Same `ServerDefault` resolution as `CreateTopic` above; rebuild - // the prepare only if a sentinel actually needs stamping, else - // project the untouched buffer zero-copy. - let needs_rewrite = request.max_topic_size == 0 || request.message_expiry == 0; - if request.max_topic_size == 0 { - request.max_topic_size = self.default_max_topic_size.get(); - } - if request.message_expiry == 0 { - request.message_expiry = self.default_message_expiry.get(); - } - if needs_rewrite { - let body = request.to_bytes(); - return Ok(build_prepare_message( - consensus, - &header, - Operation::UpdateTopic, - &body, - )); - } - Ok(message.project(consensus)) - } + // `UpdateTopic` deliberately takes the default arm: unlike create, + // an update stores `ServerDefault` sentinels verbatim (legacy + // parity), so a later get echoes `ServerDefault` instead of the + // node default frozen at update time. _ => Ok(message.project(consensus)), } } @@ -3735,9 +3702,9 @@ where // `created_at` on every CreateStream/CreateTopic/CreatePartitions. The // in-process callers that bypass `Project::project` build their prepare // through this helper directly (the CreateTopic/CreatePartitions - // assignment rewrites, the UpdateTopic default-size rewrite, and the - // PAT-cleaner delete); the stamp is load-bearing for the creates and inert - // for the UpdateTopic rewrite and the delete, whose applies ignore it. + // assignment rewrites and the PAT-cleaner delete); the stamp is + // load-bearing for the creates and inert for the delete, whose apply + // ignores it. // Shared `next_monotonic_timestamp` keeps the in-process path on the same // monotonic-clock guard as the wire path. let timestamp = consensus.next_monotonic_timestamp(); @@ -3762,9 +3729,9 @@ where // Carry the acting user id so the in-apply RBAC gate sees the same // identity on every replica. The default projection copies it (see // `Project::project`); this helper builds prepares for the ops it - // rewrites (the CreateTopic/CreatePartitions assignment rewrites, the - // UpdateTopic default-size rewrite, and the PAT-cleaner delete), which - // would otherwise reset it to 0 via `..Default::default()`. + // rewrites (the CreateTopic/CreatePartitions assignment rewrites and + // the PAT-cleaner delete), which would otherwise reset it to 0 via + // `..Default::default()`. user_id: request.user_id, // Seal the body integrity field over the rewritten body, exactly as // `Project::project` does for wire-projected prepares. This helper builds diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index dad3f70e3..63b40c20f 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -20,7 +20,10 @@ use crate::journal::{MessageLookup, PartitionJournal, PartitionJournalMemStorage use crate::log::JournalInfo; use crate::log::SegmentedLog; use crate::messages_writer::MessagesWriter; -use crate::offset_storage::{delete_persisted_offset, persist_offset, persist_offset_max}; +use crate::offset_storage::{ + PURGE_GENERATION_FILE, delete_persisted_offset, persist_offset, persist_offset_max, + persist_purge_generation, read_purge_generation, +}; use crate::poll_plan::{ AutoCommitCtx, AutoCommitTarget, DiskReadPlan, DiskSegment, LastPolledCtx, PartitionDirResolution, PollPlan, PollTier, ResidentTailSnapshot, @@ -158,6 +161,15 @@ 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, + /// 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, + /// repair and retransmission) while wiping the segments, so without the + /// floor a pre-purge op committing after the purge would flush purged + /// bytes back into a fresh segment or re-advance the reset offset. Not + /// persisted: the in-memory journal dies with the process, so no resident + /// pre-purge entry survives a restart. + purge_floor_op: u64, /// Durable superblock for this partition's consensus group, recording /// `(view, log_view)` across a crash so this replica can never /// re-participate in a view older than one it advertised. `None` for @@ -316,6 +328,17 @@ pub enum PurgeError { /// `ENOSPC` / `EIO` is logged by the superblock writer on the first failure /// and at every power-of-two thereafter. FrontierNotRecorded, + /// The wipe ran and the fresh chain is planted, but the applied purge + /// generation could not be recorded durably (`purge.gen`), so + /// `applied_purge_generation` stays at its pre-purge value and the + /// reconciler re-issues the purge. Retry, do not fence: the partition is + /// serviceable and re-purging an already-empty chain is cheap. + /// + /// Sets [`Self::purge_deferred`] for the same reason as + /// [`Self::FrontierNotRecorded`]: an op acked between this failure and the + /// retry would be wiped by that retry while every peer that recorded the + /// generation keeps it. + GenerationNotRecorded(IggyError), /// A step after the drain failed, so the partition holds no serviceable /// segment chain and its next append would panic on `active_segment()`. /// The caller must fence this group for rebuild. @@ -329,6 +352,11 @@ impl fmt::Display for PurgeError { f, "could not record the purge's offset-frontier reset; nothing was mutated" ), + Self::GenerationNotRecorded(source) => write!( + f, + "purge reset the partition but could not record its applied generation; \ + the purge will be re-issued: {source}" + ), Self::Unserviceable(source) => write!( f, "purge left the partition without a serviceable chain: {source}" @@ -436,6 +464,7 @@ where persisted_offsets: RefCell::new(HashMap::new()), observed_view, applied_purge_generation: 0, + purge_floor_op: 0, superblock: None, superblock_lock: LocalGate::new(), superblock_write_failures: Cell::new(0), @@ -462,6 +491,34 @@ where self.applied_purge_generation } + /// See [`Self::purge_floor_op` field docs](#structfield.purge_floor_op). + /// Exposed for the repair-serving path: a peer must not serve entries at + /// or below this replica's floor. + #[must_use] + pub const fn purge_floor_op(&self) -> u64 { + self.purge_floor_op + } + + /// 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). + /// + /// # Errors + /// Propagates a real I/O failure reading `purge.gen`: booting with the + /// sentinel 0 instead would make the reconciler silently re-purge and + /// destroy post-purge messages, so the boot fails loud. + 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?; + } + Ok(()) + } + #[must_use] pub const fn consensus(&self) -> &VsrConsensus<B> { &self.consensus @@ -2595,6 +2652,13 @@ where } continue; } + // Purge floor: a pre-purge batch committing after the + // purge must not flush its (purged) bytes into the fresh + // segment. It still counts into `chunk_len`, so it joins + // the evictable prefix and commit_min advances normally. + if peek_op(&entry) <= self.purge_floor_op { + continue; + } // Resident committed SendMessages entry: this node stamped it // in `append_messages` (recomputing the batch checksum over these // exact bytes), so a validating re-decode would only re-hash ~1 @@ -2752,7 +2816,13 @@ where } let retained = self.log.journal().inner.evict_prefix(count).await; let mut retained_info = JournalInfo::default(); - for (_, meta) in &retained { + for (entry, meta) in &retained { + // Purge floor: a retained pre-purge batch must not fold its + // accounting back into `journal.info`, or the info would re-adopt + // a pre-purge `current_offset` the purge just reset. + if peek_op(entry) <= self.purge_floor_op { + continue; + } if let Some(meta) = meta { accumulate_committed_info( &mut retained_info, @@ -2913,6 +2983,14 @@ 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 + // `commit_partition_entry` from re-advancing the reset offset + // and stats with pre-purge values. + if entry.header.op <= self.purge_floor_op { + return None; + } match self.committed_batch_stats_for_prepare(&entry.header) { Ok(batch_stats) => batch_stats, @@ -3196,6 +3274,18 @@ where let write_lock = self.write_lock.clone(); let _guard = write_lock.lock().await; + // Purge floor: the purge cleared the offset maps and files, so a + // pre-purge store committing now must not resurrect its offset. An op + // guard, not a bare staged-table clear at purge time: + // `restage_consumer_offset_from_journal` re-derives pending commits + // from the kept journal entries, so a cleared table alone would be + // repopulated from the entry this guard is fencing. + if prepare_header.op <= self.purge_floor_op { + self.pending_consumer_offset_commits + .remove(&prepare_header.op); + return true; + } + if let Err(error) = self .apply_staged_consumer_offset_commit(prepare_header.op) .await @@ -3883,6 +3973,48 @@ where self.stats.zero_out_all(); self.stats.increment_segments_count(1); + // Fence the resident journal instead of clearing it: entries are + // consensus history (backup commit walks, repair, retransmission), so + // they stay, but every journal-apply path no-ops ops at or below this + // floor (see `purge_floor_op`). The write lock held here is the same + // one appends take, and the pump is single-threaded, so no op can be + // assigned between reading the sequence and installing the floor. + self.purge_floor_op = self.consensus.sequencer().current_sequence(); + // The journal's flush accounting and resident poll indexes describe + // pre-purge bytes; reset them so the flush threshold counts only + // post-purge appends and polls fall back to the (fresh, empty) + // segments instead of resolving purged resident entries. + self.log.journal_mut().info = JournalInfo::default(); + self.log + .journal() + .inner + .clear_poll_index(self.purge_floor_op); + + // Last durable step: record the applied generation before the + // in-memory marker advances. On a write failure the marker stays old, + // the error propagates, and the reconciler retries the whole purge + // (idempotent, the chain is already empty). The reverse order would + // ack a purge that a crash then silently undoes: restart would + // hydrate the old generation, yet the reconciler believes the purge + // applied. Deferring PrepareOk mirrors the frontier-record failure: + // an op acked now would be wiped by the retry purge while peers that + // 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 { + self.purge_deferred = true; + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = namespace.inner(), + generation, + %error, + "purge reset the partition but could not record its applied generation; \ + deferring PrepareOk until the re-issued purge records it" + ); + return Err(PurgeError::GenerationNotRecorded(error)); + } + } self.applied_purge_generation = generation; // Same commit frontier, different (now empty) bytes: a cached offer // built pre-purge would advertise files the purge just unlinked. @@ -4242,6 +4374,17 @@ fn peek_operation(entry: &Frozen<4096>) -> Operation { .operation } +/// The consensus op of a journal entry, same cheap header cast as +/// [`peek_operation`]. Used by the purge-floor guards to tell pre-purge +/// entries (op at or below the floor) from post-purge ones. +fn peek_op(entry: &Frozen<4096>) -> u64 { + bytemuck::checked::try_from_bytes::<PrepareHeader>( + &entry[..std::mem::size_of::<PrepareHeader>()], + ) + .expect("journal entry must begin with a valid prepare header") + .op +} + /// Success reply body for a committed partition op other than `SendMessages` /// (which confirms its offsets through [`send_messages_reply_body`]). /// @@ -4416,7 +4559,7 @@ mod tests { const TEST_CLUSTER: u128 = 1; - fn test_partition() -> IggyPartition<IggyMessageBus> { + pub(super) fn test_partition() -> IggyPartition<IggyMessageBus> { let namespace = IggyNamespace::new(1, 1, 0); let consensus = VsrConsensus::new( TEST_CLUSTER, @@ -5136,7 +5279,7 @@ mod tests { /// One-message segment record in on-disk layout `[256B command header][blob]` /// stamped at `base_offset`, with a valid batch checksum so it decodes /// through `decode_batch_slice` and matches an `Offset` poll. - fn build_segment_record(namespace: IggyNamespace, base_offset: u64) -> Vec<u8> { + pub(super) fn build_segment_record(namespace: IggyNamespace, base_offset: u64) -> Vec<u8> { let mut batch = IggyMessages2::with_capacity(1); batch.push(IggyMessage2 { header: IggyMessage2Header { @@ -5834,7 +5977,7 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } - fn repair_config() -> PartitionsConfig { + pub(super) fn repair_config() -> PartitionsConfig { PartitionsConfig { messages_required_to_save: 1, size_of_messages_required_to_save: IggyByteSize::from(1024 * 1024), @@ -6123,8 +6266,8 @@ mod tests { // A purge at the origin is the one legitimate rewind, and the artifact // carries the generation that proves it: the same offer passes the fence // once its generation advances past the COMMITTED one the caller reads - // off the metadata plane (0 here), not past this replica's memory-only - // applied value. + // off the metadata plane (0 here), not past this replica's applied + // value, whose `purge.gen` hydration a kill-before-record leaves stale. let purged = crate::state_transfer::ConsumerOffsetsWire { purge_generation: 1, next_offset: 0, @@ -6146,9 +6289,10 @@ mod tests { } /// The canonical post-restart rejoin: this replica applied a purge before - /// the restart, so the metadata plane's COMMITTED generation is 1 while its - /// own memory-only `applied_purge_generation` is back at 0. Gated on the - /// local field, `offered(1) > applied(0)` reads as an advancing purge and + /// the restart but was killed before the purge's `purge.gen` record step, + /// so the metadata plane's COMMITTED generation is 1 while its own + /// hydrated `applied_purge_generation` is back at 0. Gated on the local + /// field, `offered(1) > applied(0)` reads as an advancing purge and /// disables the rewind refusal -- on the one path it exists to guard. #[compio::test] async fn given_restarted_replica_when_offer_matches_committed_purge_should_refuse_rewind() { @@ -6160,7 +6304,7 @@ mod tests { assert_eq!( partition.applied_purge_generation(), 0, - "the local generation is memory-only and starts over after a restart" + "with no purge.gen record the hydrated generation starts at 0" ); let offer = crate::state_transfer::ConsumerOffsetsWire { @@ -6399,3 +6543,296 @@ mod retention_tests { assert_eq!(nth_oldest_sealed_end(&segments, 1), None); } } + +#[cfg(test)] +mod purge_floor_tests { + use super::tests::{build_segment_record, repair_config, test_partition}; + use super::*; + use iggy_binary_protocol::{Command2, WireConsumer, WireEncode}; + + /// Fresh temp dir wired as the partition dir, so `purge()` can recreate + /// real segment files and write `purge.gen`. + fn purge_test_partition(tag: &str) -> (IggyPartition<IggyMessageBus>, std::path::PathBuf) { + let dir = std::env::temp_dir().join(format!( + "iggy-purge-floor-{tag}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + std::fs::create_dir_all(&dir).expect("create temp partition dir"); + let mut partition = test_partition(); + partition.set_partition_dir(dir.to_string_lossy().into_owned()); + (partition, dir) + } + + /// A one-message `SendMessages` prepare for `op`, journaled through the + /// replicated-apply path (which stamps offsets and re-checksums), with the + /// sequencer advanced the way `on_replicate` does after a real append. + async fn journal_send_batch(partition: &mut IggyPartition<IggyMessageBus>, op: u64) { + let namespace = IggyNamespace::new(1, 1, 0); + let record = build_segment_record(namespace, 0); + 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 = op; + header.timestamp = op; + header.namespace = namespace.inner(); + header.size = u32::try_from(total).expect("prepare size fits u32"); + }); + partition + .apply_replicated_operation(message) + .await + .expect("journal send batch"); + partition.consensus().sequencer().set_sequence(op); + } + + /// A `StoreConsumerOffset2` prepare for `op`, journaled and staged through + /// the replicated-apply path. + async fn journal_store_offset( + partition: &mut IggyPartition<IggyMessageBus>, + op: u64, + consumer_id: u32, + offset: u64, + ) { + let body = StoreConsumerOffset2Request { + consumer: WireConsumer::consumer(WireIdentifier::Numeric(consumer_id)), + stream_id: WireIdentifier::Numeric(1), + topic_id: WireIdentifier::Numeric(1), + partition_id: Some(0), + offset, + ack: AckLevel::Quorum, + } + .to_bytes(); + let header_size = std::mem::size_of::<PrepareHeader>(); + let total = header_size + body.len(); + let mut message = Message::<PrepareHeader>::new(total); + message.as_mut_slice()[header_size..].copy_from_slice(&body); + let message = message.transmute_header(|_, header: &mut PrepareHeader| { + header.command = Command2::Prepare; + header.operation = Operation::StoreConsumerOffset2; + header.op = op; + header.namespace = IggyNamespace::new(1, 1, 0).inner(); + header.size = u32::try_from(total).expect("prepare size fits u32"); + }); + partition + .apply_replicated_operation(message) + .await + .expect("journal store offset"); + partition.consensus().sequencer().set_sequence(op); + } + + #[compio::test] + async fn given_resident_batches_when_purged_should_seal_journal_polls() { + let (mut partition, dir) = purge_test_partition("seal"); + journal_send_batch(&mut partition, 1).await; + journal_send_batch(&mut partition, 2).await; + assert!( + partition + .log + .journal() + .inner + .oldest_resident_offset() + .is_some(), + "resident batches must be poll-resolvable before the purge" + ); + + partition + .purge(&repair_config(), 1) + .await + .expect("purge partition"); + + assert_eq!( + partition.log.journal().inner.oldest_resident_offset(), + None, + "purge must seal the resident poll tier so polls fall back to \ + the (fresh, empty) segments" + ); + assert_eq!( + partition.log.journal().inner.resident_entries().len(), + 2, + "journal entries are consensus history and must survive the purge" + ); + assert!( + partition.log.journal().inner.header_by_op(1).is_some() + && partition.log.journal().inner.header_by_op(2).is_some(), + "repair and retransmission must still resolve pre-purge ops" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn given_pre_purge_ops_committed_after_purge_should_advance_commit_min_without_stale_flush() + { + let (mut partition, dir) = purge_test_partition("no-stale-flush"); + journal_send_batch(&mut partition, 1).await; + journal_send_batch(&mut partition, 2).await; + + partition + .purge(&repair_config(), 1) + .await + .expect("purge partition"); + + // Both sends commit only now, after the purge fenced them. + partition.consensus().advance_commit_max(2); + partition.commit_journal(&repair_config()).await; + + assert_eq!( + partition.consensus().commit_min(), + 2, + "pre-purge ops must still commit (no wedge), just without effect" + ); + assert_eq!( + partition.offset.load(Ordering::Acquire), + 0, + "purged sends must not re-advance the reset offset" + ); + assert_eq!( + partition.log.active_segment().size.as_bytes_u64(), + 0, + "purged sends must not flush bytes into the fresh segment" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn given_post_purge_appends_when_committed_should_flush_from_offset_zero() { + let (mut partition, dir) = purge_test_partition("post-appends"); + journal_send_batch(&mut partition, 1).await; + + partition + .purge(&repair_config(), 1) + .await + .expect("purge partition"); + + // A fresh append lands after the purge; its commit walks the journal + // front where the fenced pre-purge entry still sits. + journal_send_batch(&mut partition, 2).await; + partition.consensus().advance_commit_max(2); + partition.commit_journal(&repair_config()).await; + + assert_eq!(partition.consensus().commit_min(), 2); + assert_eq!( + partition.offset.load(Ordering::Acquire), + 0, + "the single post-purge message flushes at offset 0" + ); + // Exactly ONE record's bytes: both entries stamp base_offset 0 (the + // pre-purge append was first, the post-purge one restarts at 0), so + // an unfenced flush of the purged batch would double the size while + // leaving every offset assert green. + let one_record = build_segment_record(IggyNamespace::new(1, 1, 0), 0).len() as u64; + assert_eq!( + partition.log.active_segment().size.as_bytes_u64(), + one_record, + "only the post-purge batch may reach the fresh segment" + ); + assert_eq!( + partition.log.active_segment().start_offset, + 0, + "post-purge storage restarts at offset 0" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn given_pre_purge_consumer_offset_op_when_committed_after_purge_should_not_resurrect_offset() + { + let (mut partition, dir) = purge_test_partition("offset-resurrect"); + journal_store_offset(&mut partition, 1, 7, 42).await; + + partition + .purge(&repair_config(), 1) + .await + .expect("purge partition"); + + partition.consensus().advance_commit_max(1); + partition.commit_journal(&repair_config()).await; + + assert_eq!( + partition.consensus().commit_min(), + 1, + "the fenced offset op must still commit" + ); + assert!( + partition.consumer_offsets.pin().is_empty(), + "a pre-purge store committing after the purge must not resurrect \ + the cleared consumer offset" + ); + assert!( + partition.pending_consumer_offset_commits.is_empty(), + "the fenced op must not linger in the staged-commit table" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn given_purged_straggler_when_evicted_prefix_reappends_it_should_stay_poll_sealed() { + // A flush evicts the committed prefix and re-appends the retained + // tail (`evict_prefix`); without the poll floor that re-append + // re-indexes a fenced pre-purge straggler, and resident polls serve + // purged bytes once it commits. + let (mut partition, dir) = purge_test_partition("evict-reappend"); + journal_send_batch(&mut partition, 1).await; + journal_send_batch(&mut partition, 2).await; + partition.consensus().advance_commit_max(1); + + partition + .purge(&repair_config(), 1) + .await + .expect("purge partition"); + + // The straggler flush path: evict the committed prefix (op 1), which + // re-appends the retained op 2 through `append_with_meta`. + let committed = partition.log.journal().inner.committed_prefix(1); + assert_eq!(committed.len(), 1, "only op 1 is committed"); + partition.log.journal().inner.evict_prefix(1).await; + + assert_eq!( + partition.log.journal().inner.oldest_resident_offset(), + None, + "evict re-append must not undo the purge's poll seal" + ); + assert!( + partition.log.journal().inner.header_by_op(2).is_some(), + "the retained op stays consensus history" + ); + + 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"); + partition + .purge(&repair_config(), 3) + .await + .expect("purge partition"); + assert_eq!(partition.applied_purge_generation(), 3); + + // A rebuilt partition over the same dir (restart) reads the durable + // generation instead of resetting to 0 and re-wiping. + let mut rebuilt = test_partition(); + rebuilt.set_partition_dir(dir.to_string_lossy().into_owned()); + rebuilt + .hydrate_applied_purge_generation() + .await + .expect("hydrate purge generation"); + assert_eq!( + rebuilt.applied_purge_generation(), + 3, + "restart must hydrate the durably applied purge generation" + ); + + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index c3a5b9657..f48f71a1d 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -172,7 +172,7 @@ where op_to_storage_offset: UnsafeCell<BTreeMap<u64, usize>>, /// Maps message offset -> op (for queryable entries) offset_to_op: UnsafeCell<BTreeMap<u64, u64>>, - /// Maps `(origin_timestamp, op)` -> op (for queryable entries). + /// Maps `(base_timestamp, op)` -> op (for queryable entries). /// /// Keeping `op` in the key preserves duplicate timestamps while still /// letting us seek to the closest batch for timestamp-based polling. @@ -198,6 +198,13 @@ where /// Single-replica groups have nobody to repair; retaining evicted /// entries for them is pure memory waste. repair_retention: Cell<bool>, + /// Poll-index seal installed by a partition purge: ops at or below this + /// floor never enter `offset_to_op` / `timestamp_to_op`. Without it, + /// `evict_prefix` re-appending the retained tail would re-insert + /// pre-purge entries the purge just sealed off, and resident polls would + /// serve purged bytes. Survives only as long as the journal (in-memory), + /// same lifetime argument as the partition's `purge_floor_op`. + poll_floor: Cell<u64>, } /// How many evicted entries each partition retains for repair. Sized to @@ -228,6 +235,7 @@ where evicted_ring_capacity: Cell::new(EVICTED_RING_CAPACITY), evicted_ring_bytes_max: Cell::new(EVICTED_RING_BYTES_MAX), repair_retention: Cell::new(true), + poll_floor: Cell::new(0), } } } @@ -569,9 +577,11 @@ impl PartitionJournal<PartitionJournalMemStorage> { let header = *bytemuck::checked::try_from_bytes::<PrepareHeader>(header_bytes) .expect("partition journal append expects a valid prepare header"); let op = header.op; - // One decode feeds both the offset/timestamp index (keyed on - // `origin_timestamp`) and the surfaced accounting meta (`base_timestamp`, - // size, count); the two timestamps are distinct fields, do not conflate. + // One decode feeds both the offset/timestamp index and the surfaced + // accounting meta. Both are keyed on `base_timestamp`, the broker + // append time stamped into replies: the seek hint must live on the + // same clock as `select_batch_slice`'s filter or timestamp polls seek + // to the wrong resident entry. // Trusted (no batch-hash): every entry reaching append was just stamped // by `stamp_prepare_for_persistence` (its checksum recomputed over this // exact blob) or re-appended from an already-validated resident entry, @@ -587,7 +597,7 @@ impl PartitionJournal<PartitionJournalMemStorage> { message_count, }; ( - Some((batch.header.base_offset, batch.header.origin_timestamp)), + Some((batch.header.base_offset, batch.header.base_timestamp)), Some(meta), ) } @@ -614,7 +624,13 @@ impl PartitionJournal<PartitionJournalMemStorage> { op_to_storage_offset.insert(op, storage_offset); } - if let Some((offset, timestamp)) = index_offset_timestamp { + // Poll-index only ops above the purge floor: `op_to_storage_offset` + // above stays unconditional (consensus history for the repair and + // commit walks), but a fenced pre-purge entry re-appended by + // `evict_prefix` must not become poll-resolvable again. + if op > self.poll_floor.get() + && let Some((offset, timestamp)) = index_offset_timestamp + { let offset_to_op = unsafe { &mut *self.offset_to_op.get() }; offset_to_op.insert(offset, op); @@ -657,6 +673,7 @@ where evicted_ring_capacity: Cell::new(EVICTED_RING_CAPACITY), evicted_ring_bytes_max: Cell::new(EVICTED_RING_BYTES_MAX), repair_retention: Cell::new(true), + poll_floor: Cell::new(0), } } @@ -758,6 +775,25 @@ where offset_to_op.keys().next().copied() } + /// Seal the resident poll tier: clear the offset and timestamp poll + /// indexes ONLY, so `oldest_resident_offset` reads `None` and every poll + /// falls back to the on-disk segments. Called by a partition purge, which + /// wipes the segments but must KEEP the journal entries themselves: + /// headers, storage, `op_to_storage_offset` and the evicted ring are + /// consensus history that backups, repair and retransmission still walk. + /// Clearing those would wedge `commit_min` until a view change. + /// + /// `floor` (the purge's fence op) makes the seal survive eviction: + /// `evict_prefix` re-appends the retained tail, and without the floor + /// that re-append would re-index the pre-purge entries just cleared. + pub fn clear_poll_index(&self, floor: u64) { + let offset_to_op = unsafe { &mut *self.offset_to_op.get() }; + offset_to_op.clear(); + let timestamp_to_op = unsafe { &mut *self.timestamp_to_op.get() }; + timestamp_to_op.clear(); + self.poll_floor.set(floor); + } + fn candidate_start_op(&self, query: &MessageLookup) -> Option<u64> { match query { MessageLookup::Offset { offset, .. } => { @@ -886,8 +922,12 @@ pub fn select_batch_slice( .. } => offset >= query_offset, MessageLookup::Timestamp { timestamp, .. } => { - batch.header.origin_timestamp + u64::from(record.message.header.timestamp_delta) - >= timestamp + // Match on the broker append time: replies stamp every message + // with the flat batch `base_timestamp` (the per-message delta + // applies to `origin_timestamp` only), so filtering on the + // producer clock would skip the message stamped exactly at the + // queried timestamp. + batch.header.base_timestamp >= timestamp } }; if !selected { @@ -1054,9 +1094,14 @@ pub fn select_resident( #[cfg(test)] mod tests { use super::*; + use bytes::Bytes; use iggy_binary_protocol::{Command2, HEADER_SIZE}; use journal::Journal; use server_common::Message; + use server_common::send_messages2::{ + IggyMessage2, IggyMessage2Header, IggyMessages2, SendMessages2Owned, decode_batch_slice, + }; + use server_common::sharding::IggyNamespace; fn build_prepare(op: u64, size: usize) -> Message<PrepareHeader> { Message::<PrepareHeader>::new(size).transmute_header(|_, h: &mut PrepareHeader| { @@ -1215,4 +1260,72 @@ mod tests { "from_op past commit_max yields nothing" ); } + + /// Three-message batch with the broker append time (`base_timestamp`) + /// deliberately AFTER every producer stamp (`origin_timestamp` + deltas), + /// the layout every real batch has (the broker stamps later than the + /// producer). Timestamp polls filter on the broker time because that is + /// the timestamp replies surface per message. + fn build_timestamped_batch(base_timestamp: u64, origin_timestamp: u64) -> Vec<u8> { + let mut messages = IggyMessages2::with_capacity(3); + for index in 0..3u64 { + messages.push(IggyMessage2 { + header: IggyMessage2Header { + origin_timestamp: origin_timestamp + index, + payload_length: 8, + ..Default::default() + }, + payload: Bytes::from_static(b"abcdefgh"), + user_headers: None, + }); + } + let mut owned = SendMessages2Owned::from_messages(IggyNamespace::new(1, 1, 0), &messages) + .expect("build send_messages batch"); + owned.header.base_timestamp = base_timestamp; + owned.header.batch_checksum = owned.header.checksum_for_blob(&owned.blob); + + let mut record = vec![0u8; COMMAND_HEADER_SIZE + owned.blob.len()]; + owned.header.encode_into(&mut record[..COMMAND_HEADER_SIZE]); + record[COMMAND_HEADER_SIZE..].copy_from_slice(&owned.blob); + record + } + + #[test] + fn timestamp_poll_at_exact_broker_timestamp_includes_the_batch() { + // A client polls with a timestamp read from a previous reply, which is + // the batch `base_timestamp`. Filtering on the producer origin clock + // (always a little earlier) made `origin >= base` false and silently + // skipped the message stamped exactly at the queried time. + let record = build_timestamped_batch(1_000, 900); + let batch = decode_batch_slice(&record).expect("batch decodes"); + + let at_exact = select_batch_slice( + &batch, + MessageLookup::Timestamp { + timestamp: 1_000, + count: 10, + ceiling: u64::MAX, + }, + 0, + ) + .expect("selection at the exact broker timestamp"); + assert_eq!( + at_exact.matched_messages, 3, + "poll at the reported timestamp must include the whole batch" + ); + + assert!( + select_batch_slice( + &batch, + MessageLookup::Timestamp { + timestamp: 1_001, + count: 10, + ceiling: u64::MAX, + }, + 0, + ) + .is_none(), + "poll past the broker timestamp must match nothing" + ); + } } diff --git a/core/partitions/src/offset_storage.rs b/core/partitions/src/offset_storage.rs index 393843544..1e357c988 100644 --- a/core/partitions/src/offset_storage.rs +++ b/core/partitions/src/offset_storage.rs @@ -24,6 +24,10 @@ use std::path::Path; 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). +pub const PURGE_GENERATION_FILE: &str = "purge.gen"; + 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 @@ -83,6 +87,35 @@ 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. +/// +/// # 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 +} + +/// 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 +/// 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) + .await + .map(|offset| offset.unwrap_or(0)) +} + /// Read a single persisted consumer offset. `None` if the file is absent or /// torn (shorter than 8 bytes): a crash between `persist_offset`'s truncate /// and write leaves a short file, and the boot-time loader already skips such @@ -197,6 +230,48 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[compio::test] + async fn purge_generation_absent_or_torn_is_zero_but_io_error_propagates() { + let dir = unique_temp_dir(); + let path = dir + .join(PURGE_GENERATION_FILE) + .to_string_lossy() + .into_owned(); + + assert_eq!( + read_purge_generation(&path).await.expect("absent file"), + 0, + "absent file is 0" + ); + + persist_purge_generation(&path, 3) + .await + .expect("persist generation"); + assert_eq!( + read_purge_generation(&path).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"), + 0, + "torn file degrades to 0 so the reconciler re-applies the purge" + ); + + // 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; + assert!( + matches!(result, Err(IggyError::CannotReadConsumerOffsets(_))), + "real I/O error must propagate, got {result:?}", + ); + + 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 833f07aba..54ba925f3 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -28,7 +28,9 @@ //! partition-specific payload handling on either end. use crate::messages_writer::MessagesWriter; -use crate::offset_storage::{delete_persisted_offset, persist_offset}; +use crate::offset_storage::{ + PURGE_GENERATION_FILE, delete_persisted_offset, persist_offset, persist_purge_generation, +}; use crate::segment::Segment; use crate::types::PartitionsConfig; use crate::{IggyIndexWriter, IggyPartition}; @@ -1977,13 +1979,13 @@ where // proves one happened. // Against the METADATA plane's committed generation, which the caller // reads off durable state, NOT against `self.applied_purge_generation`: - // that one is memory-only and reads 0 after every restart, so a - // post-restart rejoin of any ever-purged topic would see - // `offered > 0 == applied` and call it an advancing purge. That is the - // canonical rejoin, and treating it as a purge disables the - // `OfferRewindsDurableData` refusal below -- the one guard standing - // between an offer that rewinds this replica's offset space and its - // durable data. + // that one hydrates from `purge.gen`, which a kill before the purge's + // record step leaves absent or stale, so a post-restart rejoin of an + // ever-purged topic could see `offered > applied` and call it an + // advancing purge. That is the canonical rejoin, and treating it as a + // purge disables the `OfferRewindsDurableData` refusal below -- the + // one guard standing between an offer that rewinds this replica's + // offset space and its durable data. let purge_advances = offsets_wire.purge_generation > committed_purge_generation; let local_next_offset = self.offset_frontier(); if !purge_advances && local_next_offset > 0 && offsets_wire.next_offset < local_next_offset @@ -2552,7 +2554,33 @@ where self.stats.set_current_offset(end); // A receiver that missed a purge must not be re-wiped by the - // reconciler right after installing post-purge data. + // reconciler right after installing post-purge data. Recorded durably + // for the same reason the purge itself records it: the reconciler + // gate hydrates from `purge.gen` at boot, so a memory-only stamp + // would make a restart re-purge the just-installed data and pull it + // all over again. A write failure only re-opens that restart window + // (the wipe-then-retransfer is self-healing, peers keep the data), + // so it degrades like the offset writes above instead of failing the + // install. + if offsets_wire.purge_generation > self.applied_purge_generation + && 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 + { + tracing::warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.consensus().namespace(), + purge_generation = offsets_wire.purge_generation, + %error, + "state-transfer install could not record the offered purge \ + generation; a restart before the next purge records it will \ + re-purge and re-transfer this partition" + ); + offsets_written = false; + } + } self.applied_purge_generation = self .applied_purge_generation .max(offsets_wire.purge_generation); diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index 57994d9cc..6c2f55d0f 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -1071,7 +1071,7 @@ async fn shard_main( // instead of (0, 0). No-op on peer shards, which have no coordinator. metadata.seed_checkpoint_ref(checkpoint_seed.0, checkpoint_seed.1); // Shard 0's copy resolves the `ServerDefault` sentinels (max topic size and - // message expiry) at admission; every shard's copy backs the same resolution in responses. + // message expiry) at create admission; responses echo stored values verbatim. metadata.set_default_max_topic_size(config.system.topic.max_size.as_bytes_u64()); metadata.set_default_message_expiry(u64::from(config.system.topic.message_expiry)); // Keep the forced-checkpoint margin >= the configured prepare-queue @@ -2359,6 +2359,7 @@ async fn load_partition( config.partition.evicted_ring_bytes_max.as_bytes_u64(), ); partition.set_partition_dir(partition_dir); + partition.hydrate_applied_purge_generation().await?; hydrate_partition_log( &mut partition, config, @@ -3614,6 +3615,15 @@ fn make_metadata_commit_notifier( /// before journaling, so a committed prepare only ever carries the /// assignment-bearing variant. Kept as defense-in-depth against a future /// commit path that emits a bare op. +/// +/// "Partition-shape" is not only the partition SET: the purge and truncate +/// ops leave the set intact but advance per-partition state (purge +/// generation, delete watermark) that only the reconciler enforces on disk. +/// Omitting them defers the on-disk effect to the periodic safety tick, +/// stretching a purge's client-visible tail to a full +/// `reconcile_periodic_interval`. `DeleteSegments` is absent by design: the +/// leader rewrites it into `TruncatePartition` before journaling, so no +/// commit ever carries it. const fn operation_triggers_partition_reconcile(op: Operation) -> bool { matches!( op, @@ -3624,6 +3634,9 @@ const fn operation_triggers_partition_reconcile(op: Operation) -> bool { | Operation::DeleteTopic | Operation::DeleteStream | Operation::DeletePartitions + | Operation::PurgeStream + | Operation::PurgeTopic + | Operation::TruncatePartition ) } @@ -3649,6 +3662,29 @@ mod tests { ); } + #[test] + fn reconciler_driven_ops_broadcast_a_commit_tick() { + // These commit without touching the partition set, so nothing else + // signals the reconciler: `reconcile_partition_purges` and + // `reconcile_segment_truncations` are the only code that turns them + // into on-disk effect, and they run only when a pass runs. Dropping + // one from the filter silently downgrades it to the periodic tick. + for op in [ + Operation::PurgeStream, + Operation::PurgeTopic, + Operation::TruncatePartition, + ] { + assert!( + operation_triggers_partition_reconcile(op), + "{op:?} is enforced by the reconciler and must wake it on commit" + ); + } + assert!( + !operation_triggers_partition_reconcile(Operation::CreateUser), + "ops with no partition-shape effect must stay off the broadcast" + ); + } + #[test] fn recovery_barrier_deadline_holds_the_floor_for_small_heartbeats() { // Below the 5s default the heartbeat-independent recovery term (~7s of diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index cbae3f257..5769d0e99 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -69,9 +69,13 @@ use iggy_binary_protocol::requests::consumer_offsets::{ GetConsumerOffsetRequest, StoreConsumerOffset2Request, }; use iggy_binary_protocol::requests::messages::PollMessagesRequest; +use iggy_binary_protocol::requests::partitions::{ + CreatePartitionsRequest, DeletePartitionsRequest, +}; use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; use iggy_binary_protocol::requests::system::get_client::GetClientRequest; use iggy_binary_protocol::requests::system::get_snapshot::GetSnapshotRequest; +use iggy_binary_protocol::requests::topics::CreateTopicRequest; use iggy_binary_protocol::requests::users::{LoginRegisterRequest, LoginRegisterWithPatRequest}; use iggy_binary_protocol::responses::clients::client_response::ConsumerGroupInfoResponse; use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse; @@ -80,10 +84,12 @@ use iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse; use iggy_binary_protocol::responses::system::get_snapshot::GetSnapshotResponse; use iggy_binary_protocol::{ AckLevel, ClientVersionInfo, Command2, EvictionReason, GenericHeader, HEADER_SIZE, - KIND_CONSUMER_GROUP, Operation, ProtocolVersion, RequestHeader, WireDecode, WireEncode, - WireIdentifier, is_protocol_compatible, + KIND_CONSUMER_GROUP, MAX_PARTITIONS_PER_REQUEST, Operation, ProtocolVersion, RequestHeader, + WireDecode, WireEncode, WireIdentifier, is_protocol_compatible, +}; +use iggy_common::{ + IggyError, MaxTopicSize, PollingStrategy, SnapshotCompression, SystemSnapshotType, }; -use iggy_common::{IggyError, PollingStrategy, SnapshotCompression, SystemSnapshotType}; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; use message_bus::AUTO_COMMIT_CLIENT_ID; @@ -734,6 +740,81 @@ fn pop_next_client_request( message } +/// 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). +pub(crate) const fn validate_partitions_count(partitions_count: u32) -> Result<(), IggyError> { + if partitions_count > MAX_PARTITIONS_PER_REQUEST { + return Err(IggyError::TooManyPartitions); + } + Ok(()) +} + +/// 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. +/// `ServerDefault` is exempt from the size floor (it resolves against server +/// config at admission, matching legacy); `Unlimited` passes numerically. +pub(crate) fn validate_topic_bounds( + system_config: &NgSystemConfig, + partitions_count: u32, + max_topic_size: MaxTopicSize, +) -> Result<(), IggyError> { + validate_partitions_count(partitions_count)?; + if !matches!(max_topic_size, MaxTopicSize::ServerDefault) + && max_topic_size.as_bytes_u64() < system_config.segment.size.as_bytes_u64() + { + return Err(IggyError::InvalidTopicSize( + max_topic_size, + system_config.segment.size, + )); + } + Ok(()) +} + +/// Reject a request before it reaches consensus: warn, then send the typed +/// deny reply. A silent drop would wedge every later request on the +/// connection until the socket read timeout. `context` labels the rejection +/// site in both log lines. +#[allow(clippy::future_not_send)] +async fn send_pre_consensus_deny<B, MJ, S, SB>( + shard: &Rc<ShellShard<B, MJ, S, SB>>, + header: &RequestHeader, + transport_client_id: u128, + error: &IggyError, + context: &'static str, +) where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, +{ + warn!( + transport_client_id, + error = %error, + operation = ?header.operation, + context, + "denying request pre-consensus" + ); + let commit = current_metadata_commit(shard); + let reply = build_deny_reply(header, transport_client_id, 0, commit, error.as_code()); + if let Err(send_error) = shard + .bus + .send_to_client(transport_client_id, reply.into_generic().into_frozen()) + .await + { + warn!( + transport_client_id, + error = %send_error, + context, + "failed to send pre-consensus deny reply" + ); + } +} + #[allow(clippy::future_not_send, clippy::too_many_lines)] async fn handle_client_request<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, @@ -920,29 +1001,15 @@ async fn handle_client_request<B, MJ, S, SB>( ) { Ok(rewritten) => rewritten, Err(error) => { - // Pre-consensus rejection (token cap reached, malformed body, or a - // lost session binding): deny fast with the typed code. A silent - // drop would wedge every later request on the connection until the - // socket read timeout. - warn!( + // Token cap reached, malformed body, or a lost session binding. + send_pre_consensus_deny( + shard, + &header, transport_client_id, - error = %error, - operation = ?header.operation, - "denying personal-access-token request" - ); - let commit = current_metadata_commit(shard); - let reply = build_deny_reply(&header, transport_client_id, 0, commit, error.as_code()); - if let Err(send_error) = shard - .bus - .send_to_client(transport_client_id, reply.into_generic().into_frozen()) - .await - { - warn!( - transport_client_id, - error = %send_error, - "failed to send personal-access-token deny reply" - ); - } + &error, + "personal-access-token", + ) + .await; return; } }; @@ -954,31 +1021,42 @@ async fn handle_client_request<B, MJ, S, SB>( let request = match maybe_rewrite_user_password_request(shard, request) { Ok(rewritten) => rewritten, Err(error) => { - // Malformed body: deny fast with InvalidCommand. A silent drop - // would wedge every later request on the connection until the - // socket read timeout. - warn!( - transport_client_id, - error = %error, - operation = ?header.operation, - "denying user password request" - ); - let commit = current_metadata_commit(shard); - let reply = build_deny_reply(&header, transport_client_id, 0, commit, error.as_code()); - if let Err(send_error) = shard - .bus - .send_to_client(transport_client_id, reply.into_generic().into_frozen()) - .await - { - warn!( - transport_client_id, - error = %send_error, - "failed to send password-deny reply" - ); - } + // Malformed body: deny fast with InvalidCommand. + send_pre_consensus_deny(shard, &header, transport_client_id, &error, "user-password") + .await; return; } }; + // Static bounds run pre-consensus so a rejected request burns no + // replicated log entry; HTTP covers the same bounds via + // `command.validate()`. A body that fails to decode denies typed too + // (`InvalidCommand`), instead of riding consensus just to fail there. + let bounds = match header.operation { + Operation::CreateTopic => CreateTopicRequest::decode_from(request_body(&request)) + .map_err(|_| IggyError::InvalidCommand) + .and_then(|create_topic| { + validate_topic_bounds( + system_config, + create_topic.partitions_count, + MaxTopicSize::from(create_topic.max_topic_size), + ) + }), + Operation::CreatePartitions => CreatePartitionsRequest::decode_from(request_body(&request)) + .map_err(|_| IggyError::InvalidCommand) + .and_then(|create_partitions| { + validate_partitions_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) + }), + _ => Ok(()), + }; + if let Err(error) = bounds { + send_pre_consensus_deny(shard, &header, transport_client_id, &error, "static-bounds").await; + return; + } // Enrich consumer-group Join/Leave with the client's VSR id (+ topic // partition count for Join) before replication; see `crate::consumer_group`. let request = match maybe_rewrite_consumer_group_request(shard, request).await { @@ -1828,6 +1906,19 @@ async fn handle_poll_messages<B, MJ, S, SB>( } } Err(error) => { + // A partition id that does not exist in a resolvable topic is a + // client addressing error and must surface as a typed rejection, + // not an empty poll a consumer would read as end-of-partition. + if matches!(error, IggyError::PartitionNotFound(..)) { + warn!( + transport_client_id, + error = %error, + "poll_messages rejected: partition not found" + ); + send_non_replicated_deny(shard, request, transport_client_id, error.as_code()) + .await; + return; + } // A zero-byte body would panic the SDK's `PolledMessages` // decoder; reply the 16-byte empty-poll shape instead. A generation // fence (the client's cached assignment is stale after a rebalance) @@ -2910,9 +3001,7 @@ mod tests { use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment; use iggy_binary_protocol::requests::messages::SendMessagesHeader; use iggy_binary_protocol::requests::streams::CreateStreamRequest; - use iggy_binary_protocol::requests::topics::{ - CreateTopicRequest, CreateTopicWithAssignmentsRequest, - }; + use iggy_binary_protocol::requests::topics::CreateTopicWithAssignmentsRequest; use iggy_binary_protocol::{PrepareOkHeader, ReplyHeader, WireName, WirePartitioning}; use iggy_common::defaults::DEFAULT_ROOT_USER_ID; use iggy_common::variadic; @@ -3534,4 +3623,65 @@ mod tests { status so the SDK replays it instead of timing out" ); } + + #[test] + fn create_topic_bounds_deny_pre_consensus() { + let config = NgSystemConfig::default(); + let segment_size = config.segment.size.as_bytes_u64(); + assert!(segment_size > 0, "default segment size must be nonzero"); + + assert!( + validate_topic_bounds( + &config, + MAX_PARTITIONS_PER_REQUEST, + MaxTopicSize::ServerDefault + ) + .is_ok(), + "the partition cap itself is admissible" + ); + assert!( + matches!( + validate_topic_bounds( + &config, + MAX_PARTITIONS_PER_REQUEST + 1, + MaxTopicSize::ServerDefault + ), + Err(IggyError::TooManyPartitions) + ), + "one past the partition cap must deny" + ); + // ServerDefault is numerically 0 yet exempt from the segment-size + // floor: it resolves against server config, matching legacy. + assert!(validate_topic_bounds(&config, 1, MaxTopicSize::ServerDefault).is_ok()); + assert!(validate_topic_bounds(&config, 1, MaxTopicSize::Unlimited).is_ok()); + let below_floor = MaxTopicSize::Custom((segment_size - 1).into()); + assert!( + matches!( + validate_topic_bounds(&config, 1, below_floor), + Err(IggyError::InvalidTopicSize(size, floor)) + if size == below_floor && floor == config.segment.size + ), + "custom size below the segment size must deny with the bounds" + ); + let at_floor = MaxTopicSize::Custom(config.segment.size); + assert!( + validate_topic_bounds(&config, 1, at_floor).is_ok(), + "a topic exactly one segment large is admissible" + ); + } + + #[test] + fn partitions_count_cap_denies_pre_consensus() { + assert!( + validate_partitions_count(MAX_PARTITIONS_PER_REQUEST).is_ok(), + "the cap itself is admissible" + ); + assert!( + matches!( + validate_partitions_count(MAX_PARTITIONS_PER_REQUEST + 1), + Err(IggyError::TooManyPartitions) + ), + "one past the cap must deny" + ); + } } diff --git a/core/server-ng/src/http/handlers.rs b/core/server-ng/src/http/handlers.rs index 5d3dfd90a..4513dc715 100644 --- a/core/server-ng/src/http/handlers.rs +++ b/core/server-ng/src/http/handlers.rs @@ -109,7 +109,9 @@ use serde::Deserialize; use shard::{PartitionRead, PartitionReadReply}; use crate::auth::{verify_login_credentials, verify_pat_credentials}; -use crate::dispatch::{resolve_consumer_offset_request, resolve_poll_request}; +use crate::dispatch::{ + resolve_consumer_offset_request, resolve_poll_request, validate_topic_bounds, +}; use crate::http::error::{ ConsistencyQuery, CustomError, PartitionWriteError, ProduceAck, ProduceQuery, ReadError, WriteError, @@ -794,6 +796,12 @@ pub(in crate::http) async fn create_topic( let stream_id = Identifier::from_str_value(&stream_id).map_err(WriteError::Rejected)?; // Rejects empty/oversized name, partitions_count > MAX, replication_factor == Some(0). command.validate().map_err(WriteError::Rejected)?; + validate_topic_bounds( + &state.system_config, + command.partitions_count, + command.max_topic_size, + ) + .map_err(WriteError::Rejected)?; let request = CreateTopicRequest { stream_id: identifier_to_wire(&stream_id).map_err(WriteError::Rejected)?, partitions_count: command.partitions_count, diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs index 020c5155d..9edb1e571 100644 --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@ -680,6 +680,11 @@ pub async fn build_partition_fresh( config.partition.evicted_ring_bytes_max.as_bytes_u64(), ); partition.set_partition_dir(partition_dir); + // 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. + partition.hydrate_applied_purge_generation().await?; partition.created_at = IggyTimestamp::now(); partition.offset.store(0, Ordering::Release); partition.dirty_offset.store(0, Ordering::Relaxed); diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs index de8c036fc..2bca5b9dd 100644 --- a/core/server-ng/src/partition_reconciler.rs +++ b/core/server-ng/src/partition_reconciler.rs @@ -403,10 +403,10 @@ struct PassCounters { parked_reclaimed: usize, /// Purges staged this pass. Counted so the pass does not arm the /// fast-skip: the pump can DEFER a purge it could not record - /// (`PurgeError::FrontierNotRecorded`), which leaves - /// `applied_purge_generation` unmoved and bumps no revision, so an armed - /// skip would swallow the only re-issue and drop a committed `PurgeTopic` - /// on this replica for good. + /// (`PurgeError::FrontierNotRecorded` / `GenerationNotRecorded`), which + /// leaves `applied_purge_generation` unmoved and bumps no revision, so an + /// armed skip would swallow the only re-issue and drop a committed + /// `PurgeTopic` on this replica for good. purges_staged: usize, /// Rebuilds deferred until an in-flight `ConfirmRemove` drains. Counted /// so the pass does not arm the fast-skip: the pump's drop clears the @@ -1093,7 +1093,17 @@ fn reconcile_segment_truncations(ctx: &ReconcilerCtx, counters: &mut PassCounter /// Stage a `PurgePartition` reset for every owned partition whose committed /// `PurgeTopic` generation is newer than the one the local partition last /// applied. The pump re-checks the generation before wiping, so a redundant -/// pass (e.g. from an unrelated revision bump) is a no-op. +/// pass (e.g. from an unrelated revision bump) is a no-op. A staged frame +/// that the full pump inbox drops needs no upgrade here: the staged counter +/// keeps passes running and the next one restages, and the pump's generation +/// guard makes redundant frames free. +// TODO(hubcio): purge lands per replica on reconciler timing, while StartView +// journal repair re-materializes pre-purge ops byte-identical from a peer, so +// a replica can purge and then repair purged batches back in (or the reverse). +// The purge floor skews the same way even without repair: each replica reads +// it off its LOCAL sequencer at purge-apply time, so replicas fence different +// sets of in-flight sends (live divergence, not only the StartView case). +// Ordering these needs a partition-plane checkpoint barrier; deferred. fn reconcile_partition_purges(ctx: &ReconcilerCtx, counters: &mut PassCounters) { let partitions = ctx.shard.plane.partitions(); let namespaces: Vec<_> = partitions.namespaces().copied().collect(); @@ -1147,6 +1157,7 @@ mod tests { use iggy_binary_protocol::requests::streams::{CreateStreamRequest, DeleteStreamRequest}; use iggy_binary_protocol::requests::topics::{ CreateTopicRequest, CreateTopicWithAssignmentsRequest, DeleteTopicRequest, + PurgeTopicRequest, }; use iggy_binary_protocol::{ Command2, GenericHeader, Operation, PrepareHeader, ReplyHeader, RequestHeader, @@ -2095,6 +2106,74 @@ mod tests { ); } + /// A committed purge bumps `Streams::revision` exactly once, so only the + /// pass right after the commit is revision-driven. Until the pump applies + /// the wipe (it can be busy, or the purge can fail on I/O and need a + /// retry), every later pass runs only because `purges_staged` keeps the + /// pass from arming the fast-skip; dropping the counter would strand a + /// staged-but-unapplied purge until an unrelated commit. + #[compio::test] + async fn purge_pending_keeps_reconciler_passes_running_until_applied() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-purge"); + seed_topic(&mux, 2, 0, "topic-purge", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + reconcile_pass(&ctx).await; + reconcile_pass(&ctx).await; + assert!( + !reconcile_once(&ctx).await, + "the scenario must start from a converged, fast-skipping state" + ); + + // Committed purge: generation 1 > applied 0. + let purge = PurgeTopicRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + }; + shard + .plane + .metadata() + .mux_stm + .update(build_prepare(3, Operation::PurgeTopic, &purge)) + .expect("PurgeTopic apply succeeds"); + + assert!( + reconcile_once(&ctx).await, + "the purge commit bumps the revision, so the next pass runs" + ); + assert!( + reconcile_once(&ctx).await, + "an unapplied purge must keep passes running (retry surface), \ + not arm the fast-skip" + ); + + // Pump applies the wipe; the partition catches up to generation 1. + let ns = IggyNamespace::new(0, 0, 0); + let partitions_config = shard.plane.partitions().config().clone(); + shard + .plane + .partitions() + .get_mut_by_ns(&ns) + .expect("purged partition is materialised") + .purge(&partitions_config, 1) + .await + .expect("apply staged purge"); + + assert!( + reconcile_once(&ctx).await, + "the pass observing the applied purge still runs (unarmed skip)" + ); + assert!( + !reconcile_once(&ctx).await, + "once applied, the reconciler re-converges and fast-skips again" + ); + } + /// Permanent-tombstone-wedge regression: a teardown whose disk delete /// fails sets the tombstone and removes the `shards_table` row but never /// enqueues `ConfirmRemove`, so the tombstone never lifts. If the same diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs index 289654152..e9d98d40c 100644 --- a/core/server-ng/src/responses.rs +++ b/core/server-ng/src/responses.rs @@ -78,7 +78,7 @@ use iggy_binary_protocol::{ Command2, GenericHeader, IGGY_PROTOCOL_VERSION, KIND_CONSUMER_GROUP, Operation, ReplyHeader, RequestHeader, WireDecode, WireEncode, WireIdentifier, WireName, WirePartitioning, }; -use iggy_common::{EncryptorKind, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize}; +use iggy_common::{EncryptorKind, Identifier, IggyError, IggyTimestamp}; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; use metadata::impls::metadata::StreamsFrontend; @@ -462,13 +462,33 @@ where SB: SuperblockStore + 'static, { let partition_id = partition_id.ok_or(IggyError::InvalidIdentifier)?; - shard - .plane - .metadata() - .mux_stm - .streams() - .namespace_from_partition(stream_id, topic_id, partition_id) - .ok_or(IggyError::InvalidIdentifier) + let streams = shard.plane.metadata().mux_stm.streams(); + if let Some(namespace) = streams.namespace_from_partition(stream_id, topic_id, partition_id) { + return Ok(namespace); + } + // Tell a bad partition id apart from a bad stream/topic: the former is a + // typed not-found the client can act on, the latter keeps the generic + // rejection every caller already handles. + if streams.topic_partition_ids(stream_id, topic_id).is_some() { + Err(IggyError::PartitionNotFound( + partition_id as usize, + wire_identifier_for_display(topic_id), + wire_identifier_for_display(stream_id), + )) + } else { + Err(IggyError::InvalidIdentifier) + } +} + +/// Best-effort conversion for error payloads only: the wire reply carries just +/// the error code, so a failed conversion may fall back to a default without +/// changing what the client sees. +fn wire_identifier_for_display(id: &WireIdentifier) -> Identifier { + match id { + WireIdentifier::Numeric(numeric_id) => Identifier::numeric(*numeric_id), + WireIdentifier::String(name) => Identifier::named(name.as_str()), + } + .unwrap_or_default() } /// `user_id` is the authenticated caller, used only by the identity-scoped @@ -873,8 +893,6 @@ where S: 'static, SB: SuperblockStore + 'static, { - let default_max_topic_size = shard.plane.metadata().default_max_topic_size(); - let default_message_expiry = shard.plane.metadata().default_message_expiry(); shard.plane.metadata().mux_stm.streams().read(|streams| { let Some(stream_id) = resolve_stream_id(streams, stream_id) else { return Ok(None); @@ -888,9 +906,7 @@ where topics: stream .topics .iter() - .map(|(_, topic)| { - topic_header(topic, default_max_topic_size, default_message_expiry) - }) + .map(|(_, topic)| topic_header(topic)) .collect::<Result<Vec<_>, _>>()?, })) }) @@ -1008,8 +1024,6 @@ where S: 'static, SB: SuperblockStore + 'static, { - let default_max_topic_size = shard.plane.metadata().default_max_topic_size(); - let default_message_expiry = shard.plane.metadata().default_message_expiry(); shard.plane.metadata().mux_stm.streams().read(|streams| { let Some(stream_id) = resolve_stream_id(streams, stream_id) else { return Ok(None); @@ -1026,7 +1040,7 @@ where .get(topic_id) .ok_or(IggyError::InvalidIdentifier)?; Ok(Some(GetTopicResponse { - topic: topic_header(topic, default_max_topic_size, default_message_expiry)?, + topic: topic_header(topic)?, partitions: topic .partitions .iter() @@ -1047,8 +1061,6 @@ where S: 'static, SB: SuperblockStore + 'static, { - let default_max_topic_size = shard.plane.metadata().default_max_topic_size(); - let default_message_expiry = shard.plane.metadata().default_message_expiry(); shard.plane.metadata().mux_stm.streams().read(|streams| { let resolved_stream = resolve_stream_id(streams, stream_id).ok_or_else(|| stream_not_found(stream_id))?; @@ -1059,7 +1071,7 @@ where stream .topics .iter() - .map(|(_, topic)| topic_header(topic, default_max_topic_size, default_message_expiry)) + .map(|(_, topic)| topic_header(topic)) .collect::<Result<Vec<_>, _>>() .map(|topics| GetTopicsResponse { topics }) }) @@ -1153,48 +1165,19 @@ fn stream_response(stream: &metadata::stm::stream::Stream) -> Result<StreamRespo }) } -/// Resolve a topic's stored [`MaxTopicSize`] to the wire byte value, mapping -/// the `ServerDefault` sentinel to this node's configured default. Replicated -/// apply stores the sentinel verbatim so commit stays config-independent across -/// a cluster; the per-node default is applied here on read, matching the legacy -/// "what this server treats the limit as" semantics. Primary admission stamps -/// the same default before replication, so this only bites topics whose stored -/// size predates that stamping (older snapshot / WAL data). Explicit `Custom` -/// and `Unlimited` values pass through. -fn resolve_max_topic_size(max_topic_size: MaxTopicSize, default_bytes: u64) -> u64 { - match max_topic_size { - MaxTopicSize::ServerDefault => default_bytes, - resolved => resolved.as_bytes_u64(), - } -} - -/// Resolve a topic's stored [`IggyExpiry`] to the wire micros value, mapping the -/// `ServerDefault` sentinel to this node's configured default. Mirrors -/// [`resolve_max_topic_size`]: replicated apply stores the sentinel verbatim so -/// commit stays config-independent across a cluster, and the per-node default is -/// applied here on read. Primary admission stamps the same default before -/// replication, so this only bites topics whose stored expiry predates that -/// stamping (older snapshot / WAL data). Explicit durations and never-expire pass -/// through. -fn resolve_message_expiry(message_expiry: IggyExpiry, default_micros: u64) -> u64 { - match message_expiry { - IggyExpiry::ServerDefault => default_micros, - resolved => u64::from(resolved), - } -} - -fn topic_header( - topic: &metadata::stm::stream::Topic, - default_max_topic_size: u64, - default_message_expiry: u64, -) -> Result<StreamTopicHeader, IggyError> { +/// Stored `message_expiry` and `max_topic_size` echo verbatim, `ServerDefault` +/// as the wire sentinel (0), matching legacy: create admission resolves the +/// sentinels against server config before replication, so a stored sentinel +/// came from an update and must read back as `ServerDefault`, not as the node +/// default frozen at read time. +fn topic_header(topic: &metadata::stm::stream::Topic) -> Result<StreamTopicHeader, IggyError> { Ok(StreamTopicHeader { id: usize_to_u32(topic.id)?, created_at: topic.created_at.as_micros(), partitions_count: usize_to_u32(topic.partitions.len())?, - message_expiry: resolve_message_expiry(topic.message_expiry, default_message_expiry), + message_expiry: u64::from(topic.message_expiry), compression_algorithm: topic.compression_algorithm.as_code(), - max_topic_size: resolve_max_topic_size(topic.max_topic_size, default_max_topic_size), + max_topic_size: topic.max_topic_size.as_bytes_u64(), replication_factor: topic.replication_factor, size_bytes: topic.stats.size_bytes_inconsistent(), messages_count: topic.stats.messages_count_inconsistent(), @@ -1212,13 +1195,18 @@ fn partition_response( // 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. + // + // 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". let stats = streams .stats_registry .partition_get(stream_id, topic_id, partition.id); let (segments_count, current_offset, size_bytes, messages_count) = - stats.map_or((0, 0, 0, 0), |stats| { + stats.map_or((1, 0, 0, 0), |stats| { ( - stats.segments_count_inconsistent(), + stats.segments_count_inconsistent().max(1), stats.current_offset(), stats.size_bytes_inconsistent(), stats.messages_count_inconsistent(), @@ -1741,15 +1729,12 @@ mod tests { } #[test] - fn topic_header_resolves_server_default_and_passes_explicit_values_through() { + fn topic_header_echoes_stored_size_and_expiry_verbatim() { use iggy_common::{ - CompressionAlgorithm, IggyDuration, IggyExpiry, StreamStats, TopicStats, + CompressionAlgorithm, IggyDuration, IggyExpiry, MaxTopicSize, StreamStats, TopicStats, }; use std::sync::atomic::AtomicUsize; - const NODE_DEFAULT: u64 = 4 * 1024 * 1024 * 1024; - const EXPIRY_DEFAULT_MICROS: u64 = 3_600_000_000; - let parent = Arc::new(StreamStats::default()); let topic_with = |max_topic_size, message_expiry| metadata::stm::stream::Topic { id: 0, @@ -1767,35 +1752,29 @@ mod tests { next_consumer_group_id: 0, }; - // ServerDefault (0 on the wire) resolves to this node's configured - // default for both size and expiry, not the raw 0 the pre-fix read path - // echoed. - let resolved = topic_header( - &topic_with(MaxTopicSize::ServerDefault, IggyExpiry::ServerDefault), - NODE_DEFAULT, - EXPIRY_DEFAULT_MICROS, - ) + // Stored `ServerDefault` sentinels echo the wire sentinel (0) + // verbatim, so an update to `ServerDefault` reads back as + // `ServerDefault` instead of the node default frozen at read time. + let sentinel = topic_header(&topic_with( + MaxTopicSize::ServerDefault, + IggyExpiry::ServerDefault, + )) .expect("topic header builds"); - assert_eq!(resolved.max_topic_size, NODE_DEFAULT); - assert_eq!(resolved.message_expiry, EXPIRY_DEFAULT_MICROS); - - // Explicit values round-trip unchanged, independent of the node default. - let custom = topic_header( - &topic_with( - MaxTopicSize::from(1024u64), - IggyExpiry::ExpireDuration(IggyDuration::from(5_000_000u64)), - ), - NODE_DEFAULT, - EXPIRY_DEFAULT_MICROS, - ) + assert_eq!(sentinel.max_topic_size, 0); + assert_eq!(sentinel.message_expiry, 0); + + // Explicit values round-trip unchanged. + let custom = topic_header(&topic_with( + MaxTopicSize::from(1024u64), + IggyExpiry::ExpireDuration(IggyDuration::from(5_000_000u64)), + )) .expect("topic header builds"); assert_eq!(custom.max_topic_size, 1024); assert_eq!(custom.message_expiry, 5_000_000); - let unlimited = topic_header( - &topic_with(MaxTopicSize::Unlimited, IggyExpiry::NeverExpire), - NODE_DEFAULT, - EXPIRY_DEFAULT_MICROS, - ) + let unlimited = topic_header(&topic_with( + MaxTopicSize::Unlimited, + IggyExpiry::NeverExpire, + )) .expect("topic header builds"); assert_eq!(unlimited.max_topic_size, u64::MAX); assert_eq!(unlimited.message_expiry, u64::MAX); diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 5563f219c..f9bc7cf87 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -3700,6 +3700,7 @@ where Entry = Message<PrepareHeader>, Header = PrepareHeader, >, + M: StreamsFrontend, { let header = *msg.header(); let target = header.replica; @@ -3809,6 +3810,34 @@ where } let cluster = partition.consensus().cluster(); let self_id = partition.consensus().replica(); + // Purge convergence gate: while a committed purge is not yet locally + // applied, this journal still holds pre-purge entries with NO floor + // to fence them (the floor is installed by the purge itself), so + // serving now would hand a rejoiner batches the cluster purged. + // Defer instead: no RepairDone is sent, the rejoiner's stall retry + // re-asks, and the local purge (one reconciler wake away) installs + // the floor the fence below serves behind. + let namespace = IggyNamespace::from_raw(header.namespace); + 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() { + tracing::debug!( + shard = self.id, + namespace_raw = header.namespace, + committed_purge, + applied_purge = partition.applied_purge_generation(), + "deferring repair serve until the committed purge applies locally" + ); + return; + } let to_op = header.to_op.min(partition.consensus().commit_max()); // `None` means the journal holds NOTHING, not "nothing was evicted": // the partition journal is memory-only and `clear_all` wipes the @@ -3821,12 +3850,22 @@ where // journal instead reports eviction from the commit frontier, which // refuses the floor into a transfer (the empty window passes the // completeness check) and heals in one round. + // + // Purge fence on top: never serve entries at or below this replica's + // purge floor. The journal keeps them (own commit walk), but a + // rejoiner's floor died with its process, so served pre-purge batches + // would flush right back into its freshly reset segments. Reporting + // the floor as the retention start rides the normal `RangeEvicted` + // path: the rejoiner moves its commit floor to the purge point + // instead. + let purge_floor = partition.purge_floor_op(); let retained_from = partition .log .journal() .inner .repair_retained_from() - .unwrap_or_else(|| partition.consensus().commit_min().saturating_add(1)); + .unwrap_or_else(|| partition.consensus().commit_min().saturating_add(1)) + .max(purge_floor.saturating_add(1)); let mut from_op = header.from_op; if retained_from > from_op { self.send_repair_range_reply( @@ -6863,9 +6902,10 @@ where // purged data durably: the local applied value stays at the newer // generation, and the reconciler's `committed > applied` gate never // re-fires. Compared against the METADATA plane's committed value, not - // this partition's applied one -- the latter is memory-only and reads 0 - // after every restart. Routed through the ordinary failure arm, which - // rotates the peer; worst case is one wasted pull. + // this partition's applied one -- the latter hydrates from `purge.gen`, + // which a kill before the purge's record step leaves absent or stale. + // Routed through the ordinary failure arm, which rotates the peer; + // worst case is one wasted pull. let committed_purge_generation = self .plane .metadata() diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 955353200..dd97c2131 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -725,6 +725,24 @@ where the reconciler re-issues it while the generation stays unapplied" ); } + Err(error @ partitions::PurgeError::GenerationNotRecorded(_)) => { + // NOT fenced: the wipe ran and a fresh chain is + // planted, so the partition is serviceable; only + // the durable generation record failed, which + // leaves `applied_purge_generation` unmoved and + // the reconciler re-issuing the (now cheap) purge. + // Same pacing argument as the frontier deferral + // above; the caches already describe wiped bytes. + self.drop_partition_transfer_state(namespace, partition); + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + generation, + %error, + "purge-partition deferred: reset applied but the generation \ + record failed; the reconciler re-issues it" + ); + } Err(error @ partitions::PurgeError::Unserviceable(_)) => { // Past the drain, so this group has no serviceable // chain and the next append panics on
