This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch ci/cpp-vsr-only in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 1eb24cda77c932d5a680e54b05770d9585b94024 Author: Hubert Gruszecki <[email protected]> AuthorDate: Sat Aug 8 09:57:32 2026 +0200 feat(cpp): run C++ SDK tests only against the VSR server The C++ bindings were the last SDK exercised only against the legacy server. Moving the e2e and BDD lanes to a vsr-built iggy-server-ng meant rewriting every legacy-only expectation: logout drops the client-table entry, flush_unsaved_buffer denies typed, cluster metadata is readable pre-login, group ids are minted monotonically, and a send reports the written offsets. Four gaps surfaced as bugs. Stats hardcoded clients_count to 0 because the sync read path could not run the cross-shard ListClients gather. A poll against a missing stream or topic answered the generic InvalidIdentifier. An over-count partition delete acked a no-op instead of InvalidPartitionsCount. A purge acked before its counters moved, so an immediate read saw pre-purge totals; the reset now runs in the apply, gated on the purge generation because left-right replays it on the second buffer after the ack. --- .github/actions/cpp-bazel/pre-merge/action.yml | 6 + .github/workflows/_test_bdd.yml | 12 +- core/integration/tests/server/mod.rs | 3 + .../integration/tests/server/poll_semantics_vsr.rs | 68 +++- core/integration/tests/server/purge_vsr.rs | 99 +++++ core/integration/tests/server/stats_vsr.rs | 39 ++ .../tests/server/topic_admission_vsr.rs | 69 +++- core/metadata/src/stm/result.rs | 1 + core/metadata/src/stm/stream.rs | 413 ++++++++++++++++++--- core/server-ng/src/dispatch.rs | 24 +- core/server-ng/src/http/reads.rs | 14 +- core/server-ng/src/responses.rs | 41 +- core/simulator/src/workload/effect.rs | 10 + .../src/workload/ops/create_partitions.rs | 17 +- .../src/workload/ops/delete_partitions.rs | 58 ++- core/simulator/src/workload/shadow.rs | 113 ++++-- foreign/cpp/Cargo.toml | 2 +- foreign/cpp/tests/e2e/client.cpp | 91 ++--- foreign/cpp/tests/e2e/consumer_group.cpp | 44 +-- foreign/cpp/tests/e2e/message.cpp | 7 +- foreign/python/tests/test_topic.py | 13 +- foreign/python/tests/utils.py | 33 +- scripts/run-bdd-tests.sh | 6 +- 23 files changed, 960 insertions(+), 223 deletions(-) diff --git a/.github/actions/cpp-bazel/pre-merge/action.yml b/.github/actions/cpp-bazel/pre-merge/action.yml index adce77739..71dbe7810 100644 --- a/.github/actions/cpp-bazel/pre-merge/action.yml +++ b/.github/actions/cpp-bazel/pre-merge/action.yml @@ -83,6 +83,12 @@ runs: - name: Setup server for e2e tests if: inputs.task == 'e2e' uses: ./.github/actions/utils/server-start + with: + # The C++ bindings are vsr-built, so e2e runs against the vsr server. + # TODO(hubcio): change to iggy-server once legacy server is removed + # (core/server has VSR support) + cargo-bin: iggy-server-ng + cargo-features: vsr - name: Run e2e tests if: inputs.task == 'e2e' diff --git a/.github/workflows/_test_bdd.yml b/.github/workflows/_test_bdd.yml index 811b6b6e6..1b1fbf732 100644 --- a/.github/workflows/_test_bdd.yml +++ b/.github/workflows/_test_bdd.yml @@ -54,10 +54,10 @@ jobs: # The VSR lanes need the vsr feature on both the server and the CLI, # otherwise the CLI cannot frame requests for the VSR wire protocol # and the healthcheck ping fails. The Go SDK speaks only VSR, while - # the Python wheels, Java SDK, and .NET SDK are vsr-built, so those - # suites are always on this branch. + # the Python wheels, Java SDK, .NET SDK, and C++ bindings are + # vsr-built, so those suites are always on this branch. case "${{ inputs.task }}" in - bdd-rust-vsr|bdd-go|bdd-go-race|bdd-python|bdd-node-vsr|bdd-csharp|bdd-java) + bdd-rust-vsr|bdd-go|bdd-go-race|bdd-python|bdd-node-vsr|bdd-csharp|bdd-java|bdd-cpp) # TODO: change to iggy-server once legacy server is removed (core/server has VSR support) SERVER_BIN="iggy-server-ng" echo "Building the VSR server binary and CLI (--features vsr) for BDD tests..." @@ -91,12 +91,12 @@ jobs: if: startsWith(inputs.component, 'bdd-') && startsWith(inputs.task, 'bdd-') run: | # Extract SDK name from task (format: bdd-<sdk>, or bdd-<sdk>-vsr - # for an explicit vsr lane). Python, C#, and Java have no legacy - # lane, so their plain task names run vsr. + # for an explicit vsr lane). Python, C#, Java, and C++ have no + # legacy lane, so their plain task names run vsr. SDK_NAME=$(echo "${{ inputs.task }}" | sed 's/^bdd-//; s/-vsr$//') EXTRA_FLAGS=() case "${{ inputs.task }}" in - bdd-rust-vsr|bdd-python|bdd-node-vsr|bdd-csharp) + bdd-rust-vsr|bdd-python|bdd-node-vsr|bdd-csharp|bdd-cpp) EXTRA_FLAGS+=(--vsr) # TODO: change to iggy-server once legacy server is removed (core/server has VSR support) export IGGY_SERVER_NG_PATH="target/debug/iggy-server-ng" diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs index ed6ce0743..e6dbbe31d 100644 --- a/core/integration/tests/server/mod.rs +++ b/core/integration/tests/server/mod.rs @@ -33,6 +33,9 @@ mod poll_semantics_vsr; // Create-topic static bounds deny typed before consensus. #[cfg(feature = "vsr")] mod topic_admission_vsr; +// Stats aggregates the cross-shard connected-client count, not a hardcoded 0. +#[cfg(feature = "vsr")] +mod stats_vsr; // Purge durability: applied generation survives restart; journal-resident // purged batches stay fenced behind the purge floor. #[cfg(feature = "vsr")] diff --git a/core/integration/tests/server/poll_semantics_vsr.rs b/core/integration/tests/server/poll_semantics_vsr.rs index 1f48eef36..65b95aaba 100644 --- a/core/integration/tests/server/poll_semantics_vsr.rs +++ b/core/integration/tests/server/poll_semantics_vsr.rs @@ -17,8 +17,10 @@ //! Poll semantics against server-ng (vsr): a poll aimed at a partition id the //! topic does not have must surface a typed `PartitionNotFound`, not an empty -//! poll a consumer would read as end-of-partition; the same addressing error -//! on `get_consumer_offset` must not decode as "no offset stored"; and a +//! poll a consumer would read as end-of-partition; a poll whose stream or +//! topic does not resolve must surface the legacy `StreamIdNotFound` / +//! `TopicIdNotFound` the same way; the partition addressing error on +//! `get_consumer_offset` must not decode as "no offset stored"; and a //! timestamp poll must be at-or-after, including the message stamped exactly at //! the queried timestamp (the timestamp replies report per message). @@ -87,6 +89,68 @@ async fn given_missing_partition_when_polling_should_reject_partition_not_found( 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_missing_stream_when_polling_should_reject_stream_not_found(harness: &TestHarness) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + let stream_id = Identifier::from_str_value("no-such-stream").expect("stream identifier"); + let topic_id = Identifier::from_str_value("no-such-topic").expect("topic identifier"); + + let result = client + .poll_messages( + &stream_id, + &topic_id, + Some(0), + &Consumer::default(), + &PollingStrategy::offset(0), + 1, + false, + ) + .await; + + let expected = IggyError::StreamIdNotFound(Identifier::default()).as_code(); + assert!( + matches!(&result, Err(error) if error.as_code() == expected), + "polling a missing stream must surface Err(StreamIdNotFound), got {result:?}" + ); +} + +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_missing_topic_when_polling_should_reject_topic_not_found(harness: &TestHarness) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("topicless-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("topicless-stream").expect("stream identifier"); + let topic_id = Identifier::from_str_value("no-such-topic").expect("topic identifier"); + + let result = client + .poll_messages( + &stream_id, + &topic_id, + Some(0), + &Consumer::default(), + &PollingStrategy::offset(0), + 1, + false, + ) + .await; + + let expected = + IggyError::TopicIdNotFound(Identifier::default(), Identifier::default()).as_code(); + assert!( + matches!(&result, Err(error) if error.as_code() == expected), + "polling a missing topic of an existing stream must surface Err(TopicIdNotFound), \ + got {result:?}" + ); +} + /// `get_consumer_offset` answered an unknown partition with an empty body, /// which the SDK decodes as `None` - the same value a consumer that simply has /// no stored offset yet gets back, so a client could not tell a typo from a diff --git a/core/integration/tests/server/purge_vsr.rs b/core/integration/tests/server/purge_vsr.rs index 5cd802854..eec69008e 100644 --- a/core/integration/tests/server/purge_vsr.rs +++ b/core/integration/tests/server/purge_vsr.rs @@ -18,8 +18,11 @@ //! 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. +//! Plus read-your-purge: the counters a purge acks are visible to the very +//! next read, without waiting for the reconciler's on-disk reset. use crate::server::scenarios::purge_delete_scenario; +use iggy::prelude::*; use integration::iggy_harness; // Single node: the tests reason about ONE replica's on-disk state across a @@ -49,3 +52,99 @@ async fn given_journal_resident_messages_when_purged_should_not_resurface( ) { purge_delete_scenario::run_resident_purge_no_resurface(harness).await; } + +// No sleep, no poll: the purge acks on commit and the segment prune runs later +// on the reconciler, so the reset of the counters `get_topic` / `get_stream` +// read has to happen in the replicated apply. A retry loop here would pass +// against the pre-apply behavior too. +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_purged_topic_when_getting_topic_immediately_should_report_zero_stats( + harness: &TestHarness, +) { + const STREAM: &str = "purge-stats-stream"; + const TOPIC: &str = "purge-stats-topic"; + + let client = harness.tcp_root_client().await.expect("tcp root client"); + client.create_stream(STREAM).await.expect("create stream"); + let stream_id = Identifier::from_str_value(STREAM).expect("stream identifier"); + let topic_id = Identifier::from_str_value(TOPIC).expect("topic identifier"); + client + .create_topic( + &stream_id, + TOPIC, + 1, + CompressionAlgorithm::None, + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .expect("create topic"); + + let mut messages: Vec<IggyMessage> = (0..10) + .map(|index| { + IggyMessage::builder() + .payload(format!("message-{index}").into()) + .build() + .expect("build message") + }) + .collect(); + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("send messages"); + + let before = client + .get_topic(&stream_id, &topic_id) + .await + .expect("get topic before purge") + .expect("topic exists before purge"); + assert_eq!( + before.messages_count, 10, + "the send must be counted before the purge, or the assert below proves nothing" + ); + assert!(before.size.as_bytes_u64() > 0); + + client + .purge_topic(&stream_id, &topic_id) + .await + .expect("purge topic"); + + let topic = client + .get_topic(&stream_id, &topic_id) + .await + .expect("get topic after purge") + .expect("purge keeps the topic"); + assert_eq!( + topic.messages_count, 0, + "a read right after the purge ack must not report pre-purge messages" + ); + assert_eq!(topic.size.as_bytes_u64(), 0); + assert_eq!( + topic.partitions.len(), + 1, + "purge keeps the partition, it only empties it" + ); + assert_eq!(topic.partitions[0].messages_count, 0); + assert_eq!(topic.partitions[0].size.as_bytes_u64(), 0); + assert_eq!(topic.partitions[0].current_offset, 0); + + let stream = client + .get_stream(&stream_id) + .await + .expect("get stream after purge") + .expect("purge keeps the stream"); + assert_eq!( + stream.messages_count, 0, + "the stream rollup must drop with its purged topic" + ); + assert_eq!(stream.size.as_bytes_u64(), 0); +} diff --git a/core/integration/tests/server/stats_vsr.rs b/core/integration/tests/server/stats_vsr.rs new file mode 100644 index 000000000..f09bc0c7b --- /dev/null +++ b/core/integration/tests/server/stats_vsr.rs @@ -0,0 +1,39 @@ +// 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. + +//! Stats against server-ng (vsr): `clients_count` must report the cross-shard +//! connected-client total gathered by the `ListClients` broadcast, not the +//! hardcoded 0 the sync single-shard read used to answer. + +use iggy::prelude::*; +use integration::iggy_harness; + +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_connected_clients_when_getting_stats_should_count_clients(harness: &TestHarness) { + let clients = harness.tcp_root_clients(2).await.expect("tcp root clients"); + + let stats = clients[0].get_stats().await.expect("get stats"); + + assert_eq!( + stats.clients_count, 2, + "stats must count both connected clients, got {}", + stats.clients_count + ); +} diff --git a/core/integration/tests/server/topic_admission_vsr.rs b/core/integration/tests/server/topic_admission_vsr.rs index 8f7c7da15..64bed733d 100644 --- a/core/integration/tests/server/topic_admission_vsr.rs +++ b/core/integration/tests/server/topic_admission_vsr.rs @@ -23,8 +23,10 @@ //! `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. Listing topics of a missing stream replies with an -//! empty list, as the legacy server does. +//! legacy wire behavior. Deleting more partitions than the topic has rejects +//! with `InvalidPartitionsCount` as a committed result instead of silently +//! acking a no-op. Listing topics of a missing stream replies with an empty +//! list, as the legacy server does. use std::str::FromStr; @@ -254,6 +256,69 @@ async fn given_out_of_bounds_partitions_count_when_mutating_should_reject_typed( ); } +#[iggy_harness( + test_client_transport = [Tcp], + server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) +)] +async fn given_over_count_when_deleting_partitions_should_reject_invalid_partitions_count( + harness: &TestHarness, +) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("over-count-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("over-count-stream").expect("stream identifier"); + create_topic_with( + &client, + &stream_id, + "over-count-topic", + 3, + MaxTopicSize::ServerDefault, + ) + .await + .expect("create topic"); + let topic_id = Identifier::from_str_value("over-count-topic").expect("topic identifier"); + + // Deleting more partitions than the topic has must reject with the legacy + // typed error, not silently no-op and ack. + let invalid_count = IggyError::InvalidPartitionsCount.as_code(); + let result = client.delete_partitions(&stream_id, &topic_id, 4).await; + assert!( + matches!(&result, Err(error) if error.as_code() == invalid_count), + "deleting 4 partitions of a 3-partition topic must deny with \ + InvalidPartitionsCount, got {result:?}" + ); + let topic = client + .get_topic(&stream_id, &topic_id) + .await + .expect("get topic") + .expect("topic exists"); + assert_eq!( + topic.partitions_count, 3, + "the rejected over-count delete must not remove any partition" + ); + + client + .delete_partitions(&stream_id, &topic_id, 3) + .await + .expect("deleting exactly the topic's partition count is accepted"); + let topic = client + .get_topic(&stream_id, &topic_id) + .await + .expect("get topic") + .expect("topic exists"); + assert_eq!(topic.partitions_count, 0, "all partitions are gone"); + + // Same rejection once the topic is already empty (any count exceeds 0). + let result = client.delete_partitions(&stream_id, &topic_id, 1).await; + assert!( + matches!(&result, Err(error) if error.as_code() == invalid_count), + "deleting from a zero-partition topic must deny with \ + InvalidPartitionsCount, got {result:?}" + ); +} + #[iggy_harness( test_client_transport = [Tcp], server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true) diff --git a/core/metadata/src/stm/result.rs b/core/metadata/src/stm/result.rs index 67a0a1383..6a610bf3a 100644 --- a/core/metadata/src/stm/result.rs +++ b/core/metadata/src/stm/result.rs @@ -183,6 +183,7 @@ result_enum!(CreatePartitionsResult { result_enum!(DeletePartitionsResult { StreamNotFound = 1009, TopicNotFound = 2010, + InvalidPartitionsCount = 2019, }); // `TruncatePartition` is the committed form of a client `DeleteSegments`; an // unresolvable target commits as a rejection so the request sequence stays diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index 0a97ddc8e..4bf63af85 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -362,7 +362,22 @@ impl Stream { pub struct StatsRegistry { streams: std::sync::Mutex<AHashMap<usize, Arc<StreamStats>>>, topics: std::sync::Mutex<AHashMap<(usize, usize), Arc<TopicStats>>>, - partitions: std::sync::Mutex<AHashMap<(usize, usize, usize), Arc<PartitionStats>>>, + partitions: std::sync::Mutex<AHashMap<(usize, usize, usize), PartitionEntry>>, +} + +/// Shared partition counters plus the purge generation they were last reset for. +#[derive(Debug)] +struct PartitionEntry { + stats: Arc<PartitionStats>, + /// Highest [`Partition::purge_generation`] this entry's counters were reset + /// for, the registry's mirror of the partition plane's + /// `applied_purge_generation` gate. + /// + /// Load-bearing: an apply runs on BOTH left-right buffers and the second run + /// is deferred to the next metadata publish, which can be long after the + /// purge acked. Counters are shared side state (one `Arc` across buffers), + /// so an ungated second reset would wipe messages sent since the purge. + purged_generation: u64, } impl StatsRegistry { @@ -407,7 +422,11 @@ impl StatsRegistry { .lock() .expect("stats registry mutex poisoned") .entry((stream_id, topic_id, partition_id)) - .or_insert_with(|| Arc::new(PartitionStats::new(parent))) + .or_insert_with(|| PartitionEntry { + stats: Arc::new(PartitionStats::new(parent)), + purged_generation: 0, + }) + .stats .clone() } @@ -426,7 +445,63 @@ impl StatsRegistry { .lock() .expect("stats registry mutex poisoned") .get(&(stream_id, topic_id, partition_id)) - .cloned() + .map(|entry| entry.stats.clone()) + } + + /// Reset the counters of every partition a purge just advanced, so a client + /// that reads right after the ack sees the purge instead of pre-purge + /// totals. The on-disk reset stays async (the reconciler resets each + /// partition on every replica once it observes the committed generation); + /// this only moves the counters to the shape that reset converges on. + /// + /// Reset, never decrement: `zero_out_all` swaps in 0 and rolls each parent + /// back by exactly what it swapped out, so a replayed purge entry over an + /// already-zeroed registry cannot underflow a parent total. The generation + /// gate on top makes the replay a no-op outright. + /// + /// The entry is created when missing so the gate is recorded even for a + /// partition this node has not materialized yet. A fresh entry holds no + /// segment, and `ensure_initial_segment` counts the one it plants, hence + /// the segment is restored only for a partition that already had storage -- + /// inventing one here would double-count against that later bump. + // The guard spans a read-modify-write of one entry (check the gate, stamp + // it, take the `Arc`), so it cannot collapse into the single chained + // expression the drop-tightening lint asks for. + #[allow(clippy::significant_drop_tightening)] + fn reset_purged_partitions( + &self, + stream_id: usize, + topic_id: usize, + parent: &Arc<TopicStats>, + partitions: &[Partition], + ) { + for partition in partitions { + // Guard dropped before the counters move: `zero_out_all` cascades a + // rollback into the parent topic and stream totals, which the + // registry map has no part in. + let stats = { + let mut entries = self + .partitions + .lock() + .expect("stats registry mutex poisoned"); + let entry = entries + .entry((stream_id, topic_id, partition.id)) + .or_insert_with(|| PartitionEntry { + stats: Arc::new(PartitionStats::new(Arc::clone(parent))), + purged_generation: 0, + }); + if entry.purged_generation >= partition.purge_generation { + continue; + } + entry.purged_generation = partition.purge_generation; + entry.stats.clone() + }; + let had_storage = stats.segments_count_inconsistent() > 0; + stats.zero_out_all(); + if had_storage { + stats.increment_segments_count(1); + } + } } fn remove_stream(&self, id: usize) { @@ -1508,28 +1583,31 @@ impl StateHandler for PurgeStreamRequest { type State = StreamsInner; fn apply(&self, state: &mut StreamsInner, _timestamp: IggyTimestamp) -> ApplyReply { // Stream purge = topic purge over every topic in the stream: advance - // each partition's monotonic purge generation and clear the delete - // watermark; every replica's reconciler observes the committed - // generation and resets the partition to a single empty segment at - // offset 0 with cleared offsets (see `PurgeTopicRequest`). Metadata - // shape stays intact. - let advanced = { - let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else { - return ApplyReply::err(PurgeStreamResult::StreamNotFound); - }; - let Some(stream) = state.items.get_mut(stream_id) else { - return ApplyReply::err(PurgeStreamResult::StreamNotFound); - }; - let mut advanced = false; - for (_, topic) in &mut stream.topics { - for partition in &mut topic.partitions { - partition.purge_generation = partition.purge_generation.wrapping_add(1); - partition.deleted_up_to_offset = 0; - advanced = true; - } - } - advanced + // each partition's monotonic purge generation, clear the delete + // watermark, and reset the partition counters; every replica's + // reconciler observes the committed generation and resets the partition + // to a single empty segment at offset 0 with cleared offsets (see + // `PurgeTopicRequest`). Metadata shape stays intact. + let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else { + return ApplyReply::err(PurgeStreamResult::StreamNotFound); }; + let Some(stream) = state.items.get_mut(stream_id) else { + return ApplyReply::err(PurgeStreamResult::StreamNotFound); + }; + let mut advanced = false; + for (topic_id, topic) in &mut stream.topics { + for partition in &mut topic.partitions { + partition.purge_generation = partition.purge_generation.wrapping_add(1); + partition.deleted_up_to_offset = 0; + advanced = true; + } + state.stats_registry.reset_purged_partitions( + stream_id, + topic_id, + &topic.stats, + &topic.partitions, + ); + } if advanced { state.revision = state.revision.wrapping_add(1); } @@ -1772,25 +1850,34 @@ impl StateHandler for PurgeTopicRequest { // offsets at 0 and drops the consumer-offset barrier that bounded the // trim, and the reconciler re-stages any nonzero watermark on every // pass -- a surviving one would delete post-purge segments. - let advanced = { - let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else { - return ApplyReply::err(PurgeTopicResult::StreamNotFound); - }; - let Some(topic_id) = state.resolve_topic_id(stream_id, &self.topic_id) else { - return ApplyReply::err(PurgeTopicResult::TopicNotFound); - }; - let Some(stream) = state.items.get_mut(stream_id) else { - return ApplyReply::err(PurgeTopicResult::StreamNotFound); - }; - let Some(topic) = stream.topics.get_mut(topic_id) else { - return ApplyReply::err(PurgeTopicResult::TopicNotFound); - }; - for partition in &mut topic.partitions { - partition.purge_generation = partition.purge_generation.wrapping_add(1); - partition.deleted_up_to_offset = 0; - } - !topic.partitions.is_empty() + // + // The shared partition counters are reset here too: they are read back + // by `get_topic` / `get_stream` on any node that applied this commit, + // and leaving them until the reconciler runs makes a purge ack followed + // by a read report pre-purge totals. + let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else { + return ApplyReply::err(PurgeTopicResult::StreamNotFound); }; + let Some(topic_id) = state.resolve_topic_id(stream_id, &self.topic_id) else { + return ApplyReply::err(PurgeTopicResult::TopicNotFound); + }; + let Some(stream) = state.items.get_mut(stream_id) else { + return ApplyReply::err(PurgeTopicResult::StreamNotFound); + }; + let Some(topic) = stream.topics.get_mut(topic_id) else { + return ApplyReply::err(PurgeTopicResult::TopicNotFound); + }; + for partition in &mut topic.partitions { + partition.purge_generation = partition.purge_generation.wrapping_add(1); + partition.deleted_up_to_offset = 0; + } + let advanced = !topic.partitions.is_empty(); + state.stats_registry.reset_purged_partitions( + stream_id, + topic_id, + &topic.stats, + &topic.partitions, + ); if advanced { state.revision = state.revision.wrapping_add(1); } @@ -1892,8 +1979,12 @@ impl StateHandler for DeletePartitionsRequest { }; let count_to_delete = self.partitions_count as usize; - let did_delete = count_to_delete > 0 && count_to_delete <= topic.partitions.len(); - if did_delete { + if count_to_delete > topic.partitions.len() { + return ApplyReply::err(DeletePartitionsResult::InvalidPartitionsCount); + } + // Zero count is rejected pre-consensus; a replayed legacy entry still + // applies as the historical ok no-op. + if count_to_delete > 0 { let retained = topic.partitions.len() - count_to_delete; topic.partitions.truncate(retained); // Members assigned the removed partitions must give them up. @@ -1903,8 +1994,6 @@ impl StateHandler for DeletePartitionsRequest { state .stats_registry .remove_partitions_from(stream_id, topic_id, retained); - } - if did_delete { state.revision = state.revision.wrapping_add(1); } ApplyReply::ok(Bytes::new()) @@ -2465,6 +2554,63 @@ mod tests { assert!(apply.body.is_empty()); } + /// Over-count deletes were acked ok as a silent no-op; they must commit the + /// legacy `InvalidPartitionsCount` rejection. Zero stays an ok no-op at the + /// apply (rejected pre-consensus; a replayed entry keeps its historical ack). + #[test] + fn given_delete_partitions_counts_when_applied_should_reject_over_count() { + let cases: &[(u32, u32, u32, usize)] = &[ + // (partitions in topic, count to delete, expected code, remaining) + ( + 3, + 4, + u32::from(DeletePartitionsResult::InvalidPartitionsCount), + 3, + ), + ( + 0, + 1, + u32::from(DeletePartitionsResult::InvalidPartitionsCount), + 0, + ), + (3, 0, 0, 3), + (3, 3, 0, 0), + (3, 2, 0, 1), + ]; + for &(partitions_count, count_to_delete, expected_code, expected_remaining) in cases { + let mut inner = StreamsInner::new(); + create_stream(&mut inner, "stream"); + let create_topic = CreateTopicWithAssignmentsRequest { + request: make_topic_request(0, partitions_count, "topic"), + partitions: (0..partitions_count) + .map(|partition_id| CreatedPartitionAssignment { + partition_id, + consensus_group_id: 1, + }) + .collect(), + }; + let _ = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::now()); + + let delete = DeletePartitionsRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partitions_count: count_to_delete, + }; + let apply = StateHandler::apply(&delete, &mut inner, IggyTimestamp::now()); + + assert_eq!( + apply.code, expected_code, + "deleting {count_to_delete} of {partitions_count} partitions" + ); + assert!(apply.body.is_empty()); + assert_eq!( + inner.items[0].topics[0].partitions.len(), + expected_remaining, + "deleting {count_to_delete} of {partitions_count} partitions" + ); + } + } + #[test] fn given_live_stream_when_apply_purge_stream_should_return_ok_with_empty_body() { let mut inner = StreamsInner::new(); @@ -2541,6 +2687,181 @@ mod tests { ); } + /// A purge acks on commit while the on-disk reset waits for the reconciler, + /// so the counters `get_topic` / `get_stream` read must move in the apply or + /// a read right after the ack reports pre-purge totals. + #[test] + fn given_counted_partition_when_apply_purge_topic_should_zero_the_scope() { + let mut inner = inner_with_registered_partition(); + let stats = inner.stats_registry.partition_get(0, 0, 0).expect("stats"); + stats.increment_segments_count(1); + stats.increment_messages_count(7); + stats.increment_size_bytes(512); + stats.set_current_offset(6); + assert_eq!( + inner.items[0].topics[0].stats.messages_count_inconsistent(), + 7, + "partition counters must roll up before the purge, or the test proves nothing" + ); + + let purge = PurgeTopicRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + }; + let apply = StateHandler::apply(&purge, &mut inner, IggyTimestamp::now()); + assert_eq!(apply.code, 0); + + assert_eq!(stats.messages_count_inconsistent(), 0); + assert_eq!(stats.size_bytes_inconsistent(), 0); + assert_eq!(stats.current_offset(), 0); + assert_eq!( + stats.segments_count_inconsistent(), + 1, + "a purged partition keeps the one empty segment the reset lands on" + ); + let topic_stats = &inner.items[0].topics[0].stats; + assert_eq!(topic_stats.messages_count_inconsistent(), 0); + assert_eq!(topic_stats.size_bytes_inconsistent(), 0); + let stream_stats = &inner.items[0].stats; + assert_eq!(stream_stats.messages_count_inconsistent(), 0); + assert_eq!(stream_stats.size_bytes_inconsistent(), 0); + } + + /// A stream purge walks every topic, so every topic's partitions must reset, + /// not just the first one. + #[test] + fn given_counted_partitions_when_apply_purge_stream_should_zero_every_topic() { + let mut inner = inner_with_registered_partition(); + let create_topic = CreateTopicWithAssignmentsRequest { + request: make_topic_request(0, 1, "metrics"), + partitions: vec![CreatedPartitionAssignment { + partition_id: 0, + consensus_group_id: 2, + }], + }; + let _ = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::now()); + let second_topic_stats = inner.items[0].topics[1].stats.clone(); + inner.stats_registry.partition(0, 1, 0, second_topic_stats); + + let counters: Vec<Arc<PartitionStats>> = (0..2) + .map(|topic_id| { + let stats = inner + .stats_registry + .partition_get(0, topic_id, 0) + .expect("stats"); + stats.increment_segments_count(1); + stats.increment_messages_count(9); + stats.increment_size_bytes(64); + stats + }) + .collect(); + assert_eq!(inner.items[0].stats.messages_count_inconsistent(), 18); + + let purge = PurgeStreamRequest { + stream_id: WireIdentifier::numeric(0), + }; + let apply = StateHandler::apply(&purge, &mut inner, IggyTimestamp::now()); + assert_eq!(apply.code, 0); + + for stats in &counters { + assert_eq!(stats.messages_count_inconsistent(), 0); + assert_eq!(stats.size_bytes_inconsistent(), 0); + assert_eq!(stats.segments_count_inconsistent(), 1); + } + assert_eq!(inner.items[0].stats.messages_count_inconsistent(), 0); + assert_eq!(inner.items[0].stats.size_bytes_inconsistent(), 0); + } + + /// The left-right buffers absorb every op twice and the second absorb is + /// deferred to the next metadata publish, which can land long after the + /// purge acked. Counters are shared side state, so the deferred replay must + /// leave post-purge traffic alone -- and must not decrement a parent total + /// it already rolled back. + #[test] + fn given_purged_buffer_when_other_buffer_replays_purge_should_keep_new_counters() { + let mut first = inner_with_registered_partition(); + let mut second = first.clone(); + let stats = first.stats_registry.partition_get(0, 0, 0).expect("stats"); + stats.increment_segments_count(1); + stats.increment_messages_count(10); + stats.increment_size_bytes(320); + + let purge = PurgeTopicRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + }; + let _ = StateHandler::apply(&purge, &mut first, IggyTimestamp::now()); + assert_eq!(stats.messages_count_inconsistent(), 0); + + // Sent after the ack, before the deferred absorb on the other buffer. + stats.increment_messages_count(4); + stats.increment_size_bytes(128); + + let _ = StateHandler::apply(&purge, &mut second, IggyTimestamp::now()); + assert_eq!( + second.items[0].topics[0].partitions[0].purge_generation, 1, + "the replay computes the same generation, so the gate is what stops it" + ); + assert_eq!( + stats.messages_count_inconsistent(), + 4, + "the deferred replay must not wipe post-purge counters" + ); + assert_eq!(stats.size_bytes_inconsistent(), 128); + let topic_stats = first.items[0].topics[0].stats.clone(); + assert_eq!( + topic_stats.messages_count_inconsistent(), + 4, + "a second rollback of the same total would underflow the parent" + ); + assert_eq!(topic_stats.size_bytes_inconsistent(), 128); + + // A genuinely new purge still resets: the gate is per generation. + let _ = StateHandler::apply(&purge, &mut first, IggyTimestamp::now()); + assert_eq!(stats.messages_count_inconsistent(), 0); + assert_eq!(topic_stats.messages_count_inconsistent(), 0); + } + + /// Boot replays the metadata WAL before any partition materializes, so the + /// purge has no counters to reset -- but it must still record the gate, or + /// the deferred second absorb wipes whatever the partition loaded since. + #[test] + fn given_unmaterialized_partition_when_apply_purge_should_gate_the_replay() { + let mut inner = StreamsInner::new(); + create_stream(&mut inner, "alpha"); + let create_topic = CreateTopicWithAssignmentsRequest { + request: make_topic_request(0, 1, "logs"), + partitions: vec![CreatedPartitionAssignment { + partition_id: 0, + consensus_group_id: 1, + }], + }; + let _ = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::now()); + let mut replay = inner.clone(); + + let purge = PurgeTopicRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + }; + let _ = StateHandler::apply(&purge, &mut inner, IggyTimestamp::now()); + + // The data plane materializes the partition afterwards and counts what + // it plants; the purge must not have invented a segment for it. + let topic_stats = inner.items[0].topics[0].stats.clone(); + let stats = inner.stats_registry.partition(0, 0, 0, topic_stats); + assert_eq!(stats.segments_count_inconsistent(), 0); + stats.increment_segments_count(1); + stats.increment_messages_count(5); + + let _ = StateHandler::apply(&purge, &mut replay, IggyTimestamp::now()); + assert_eq!( + stats.messages_count_inconsistent(), + 5, + "the gate recorded at apply must survive into the partition's entry" + ); + assert_eq!(stats.segments_count_inconsistent(), 1); + } + #[test] fn given_missing_topic_when_apply_purge_topic_should_return_topic_not_found() { let mut inner = StreamsInner::new(); diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index e9e84ac36..f03afd298 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -59,8 +59,9 @@ use consensus::{ use iggy_binary_protocol::PrepareHeader; use iggy_binary_protocol::codes::{ GET_CLIENT_CODE, GET_CLIENTS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE, - GET_ME_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, GET_SNAPSHOT_FILE_CODE, LOGIN_USER_CODE, - LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE, POLL_MESSAGES_CODE, SYNC_CONSUMER_GROUP_CODE, + GET_ME_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, GET_SNAPSHOT_FILE_CODE, GET_STATS_CODE, + LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE, POLL_MESSAGES_CODE, + SYNC_CONSUMER_GROUP_CODE, }; use iggy_binary_protocol::primitives::consumer::WireConsumer; use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy; @@ -1513,6 +1514,13 @@ async fn handle_default_non_replicated<B, MJ, S, SB>( send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await; return; } + // Stats is the one default read with an async input: the cross-shard + // connected-client gather. Run it here so the shared builder stays sync. + let clients_count = if code == GET_STATS_CODE { + u32::try_from(shard.list_all_clients().await.len()).unwrap_or(u32::MAX) + } else { + 0 + }; match build_non_replicated_response( shard, code, @@ -1520,6 +1528,7 @@ async fn handle_default_non_replicated<B, MJ, S, SB>( user_id, roster, client_ip, + clients_count, ) { Ok(response) => { let commit = current_metadata_commit(shard); @@ -1912,14 +1921,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 + // A stream, topic, or partition id that does not resolve 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(..)) { + if matches!( + error, + IggyError::PartitionNotFound(..) + | IggyError::StreamIdNotFound(_) + | IggyError::TopicIdNotFound(..) + ) { warn!( transport_client_id, error = %error, - "poll_messages rejected: partition not found" + "poll_messages rejected: target not found" ); send_non_replicated_deny(shard, request, transport_client_id, error.as_code()) .await; diff --git a/core/server-ng/src/http/reads.rs b/core/server-ng/src/http/reads.rs index 9246ad610..45bc10405 100644 --- a/core/server-ng/src/http/reads.rs +++ b/core/server-ng/src/http/reads.rs @@ -23,10 +23,12 @@ use crate::bootstrap::ServerNgShard; use bytes::Bytes; use consensus::MetadataHandle; use iggy_binary_protocol::WireIdentifier; +use iggy_binary_protocol::codes::GET_STATS_CODE; use iggy_common::wire_conversions::identifier_to_wire; use iggy_common::{Identifier, IggyError}; use metadata::impls::metadata::StreamsFrontend; use metadata::permissioner::Permissioner; +use send_wrapper::SendWrapper; use std::rc::Rc; use crate::http::error::{Consistency, ReadError}; @@ -78,8 +80,9 @@ pub(in crate::http) fn authorize_read( /// a TCP read of the same entity return byte-identical bodies. /// /// Reads never touch consensus or a VSR session: `build_non_replicated_response` -/// is a pure STM read. It is synchronous, so this helper is too - no submit -/// await, no gate, no `SendWrapper`. An absent entity surfaces as +/// is a pure STM read, with one exception - the stats read's cross-shard +/// connected-client gather, an async broadcast run here (under `SendWrapper`, +/// same as `/metrics`) before the sync builder. An absent entity surfaces as /// [`NonReplicatedResponse::Empty`], mapped to 404 here because every REST read /// whose entity can be missing shares that not-found shape. pub(in crate::http) async fn read_local( @@ -92,6 +95,12 @@ pub(in crate::http) async fn read_local( ) -> Result<Bytes, ReadError> { await_recovery_barrier(&state.shard).await?; authorize_read(state, identity, consistency, rule)?; + let clients_count = if code == GET_STATS_CODE { + u32::try_from(SendWrapper::new(state.shard.list_all_clients()).await.len()) + .unwrap_or(u32::MAX) + } else { + 0 + }; match build_non_replicated_response( &state.shard, code, @@ -99,6 +108,7 @@ pub(in crate::http) async fn read_local( Some(identity.user_id), &state.roster, identity.client_ip, + clients_count, ) .map_err(ReadError::Rejected)? { diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs index 8d79bf648..6ef179c98 100644 --- a/core/server-ng/src/responses.rs +++ b/core/server-ng/src/responses.rs @@ -466,18 +466,29 @@ where 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. + // Name the level that missed - partition, topic, or stream - with the + // legacy typed not-found, so a client can tell an addressing typo from an + // empty partition. Callers that shape their own reply (empty poll, group + // gather) treat every variant the same, so the split is reply-visible only + // where a caller denies typed. if streams.topic_partition_ids(stream_id, topic_id).is_some() { - Err(IggyError::PartitionNotFound( + return Err(IggyError::PartitionNotFound( partition_id as usize, wire_identifier_for_display(topic_id), wire_identifier_for_display(stream_id), - )) - } else { - Err(IggyError::InvalidIdentifier) + )); } + Err(streams.read(|inner| { + let Some(resolved_stream) = resolve_stream_id(inner, stream_id) else { + return stream_not_found(stream_id); + }; + if resolve_topic_id(inner, resolved_stream, topic_id).is_none() { + return topic_not_found(stream_id, topic_id); + } + // Unreachable while `topic_partition_ids` misses only on stream/topic; + // kept as the safe generic rejection should that invariant drift. + IggyError::InvalidIdentifier + })) } /// Best-effort conversion for error payloads only: the wire reply carries just @@ -496,7 +507,10 @@ fn wire_identifier_for_display(id: &WireIdentifier) -> Identifier { /// stays with the per-transport gates that run before this builder. `client_ip` /// is the caller's transport-level peer address, used only by the /// cluster-metadata read to pick each node's advertised address; `None` -/// degrades to the catch-all address. +/// degrades to the catch-all address. `clients_count` is the cross-shard +/// connected-client total, used only by the stats read: it comes from the async +/// `ListClients` scatter-gather, which this sync builder cannot run, so both +/// transport callers gather it up front (0 for every other opcode). pub(crate) fn build_non_replicated_response<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, code: u32, @@ -504,6 +518,7 @@ pub(crate) fn build_non_replicated_response<B, MJ, S, SB>( user_id: Option<u32>, roster: &ClusterRoster, client_ip: Option<IpAddr>, + clients_count: u32, ) -> Result<NonReplicatedResponse, IggyError> where B: ShellBus, @@ -517,7 +532,7 @@ where build_cluster_metadata_response(roster, shard, client_ip).to_bytes(), )), GET_STATS_CODE => Ok(NonReplicatedResponse::Bytes( - build_stats_response(shard)?.to_bytes(), + build_stats_response(shard, clients_count)?.to_bytes(), )), GET_STREAM_CODE => { let request = @@ -680,6 +695,7 @@ where fn build_stats_response<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, + clients_count: u32, ) -> Result<StatsResponse, IggyError> where B: ShellBus, @@ -766,12 +782,7 @@ where partitions_count, segments_count, messages_count, - // Connected clients are per-shard `SessionManager` state, aggregated - // across shards only by the async `ListClients` broadcast (see - // `get_clients`). This sync single-shard read can't gather it, and one - // shard's local count is a fraction of the total, so report 0 rather - // than a misleading partial. - clients_count: 0, + clients_count, consumer_groups_count, hostname: system.hostname, os_name: system.os_name, diff --git a/core/simulator/src/workload/effect.rs b/core/simulator/src/workload/effect.rs index f9adb7edd..a9d8c868c 100644 --- a/core/simulator/src/workload/effect.rs +++ b/core/simulator/src/workload/effect.rs @@ -42,6 +42,16 @@ pub enum Effect { stream: String, name: String, }, + AddPartitions { + stream: String, + topic: String, + count: u32, + }, + RemovePartitions { + stream: String, + topic: String, + count: u32, + }, AddUser { name: String, }, diff --git a/core/simulator/src/workload/ops/create_partitions.rs b/core/simulator/src/workload/ops/create_partitions.rs index 4ee8eb300..cb6ee7c0e 100644 --- a/core/simulator/src/workload/ops/create_partitions.rs +++ b/core/simulator/src/workload/ops/create_partitions.rs @@ -19,8 +19,9 @@ //! //! Targets `Ok` (live topic), `StreamNotFound` (fabricated parent stream), or //! `TopicNotFound` (live stream, fabricated topic). `InvalidPartitionsCount` -//! not targeted. Shadow tracks no partition counts, so every outcome predicts -//! `Effect::None`. +//! not targeted (only reachable through partition-id overflow). A committed +//! `Ok` grows the shadow's per-topic partition count, which +//! `delete_partitions` samples against. use iggy_binary_protocol::RequestHeader; use rand::RngExt; @@ -100,7 +101,13 @@ pub const fn classify_reply(code: u32) -> Outcome { } #[must_use] -pub const fn predicted_effect(_input: &Input, _outcome: Outcome) -> Effect { - // Shadow tracks no per-topic partition counts. - Effect::None +pub fn predicted_effect(input: &Input, outcome: Outcome) -> Effect { + match outcome { + Outcome::Ok => Effect::AddPartitions { + stream: input.stream.clone(), + topic: input.topic.clone(), + count: input.partitions_count, + }, + _ => Effect::None, + } } diff --git a/core/simulator/src/workload/ops/delete_partitions.rs b/core/simulator/src/workload/ops/delete_partitions.rs index 46ee23e73..81dc3bde0 100644 --- a/core/simulator/src/workload/ops/delete_partitions.rs +++ b/core/simulator/src/workload/ops/delete_partitions.rs @@ -17,11 +17,14 @@ //! `DeletePartitions` op. //! -//! Targets `Ok` (a live topic), `StreamNotFound` (a fabricated parent stream), -//! or `TopicNotFound` (a live stream with a fabricated topic). Partition counts -//! are not tracked in the shadow, so every outcome predicts `Effect::None`. +//! Targets `Ok` (a live topic, count within its shadow-tracked partition +//! count), `StreamNotFound` (a fabricated parent stream), `TopicNotFound` (a +//! live stream with a fabricated topic), or `InvalidPartitionsCount` (a live +//! topic, count one past its partition count - the server commits the typed +//! rejection instead of acking a silent no-op). A committed `Ok` shrinks the +//! shadow's per-topic partition count. -use iggy_binary_protocol::RequestHeader; +use iggy_binary_protocol::{MAX_PARTITIONS_PER_REQUEST, RequestHeader}; use rand::RngExt; use rand_xoshiro::Xoshiro256Plus; use server_common::Message; @@ -40,7 +43,12 @@ pub struct Input { pub partitions_count: u32, } -pub const OUTCOMES: &[Outcome] = &[Outcome::Ok, Outcome::StreamNotFound, Outcome::TopicNotFound]; +pub const OUTCOMES: &[Outcome] = &[ + Outcome::Ok, + Outcome::StreamNotFound, + Outcome::TopicNotFound, + Outcome::InvalidPartitionsCount, +]; pub fn sample( shadow: &mut Shadow, @@ -51,7 +59,15 @@ pub fn sample( match outcome { Outcome::Ok => { let (stream, topic) = shadow.pick_topic_pair(prng)?; - let partitions_count = 1 + prng.random_range(0..4u32); + let live = *shadow + .topic_partitions + .get(&(stream.clone(), topic.clone()))?; + if live == 0 { + // Any nonzero count would over-delete; that is the + // `InvalidPartitionsCount` target, not `Ok`. + return None; + } + let partitions_count = 1 + prng.random_range(0..live.min(4)); Some(Input { stream, topic, @@ -78,6 +94,24 @@ pub fn sample( partitions_count, }) } + Outcome::InvalidPartitionsCount => { + let (stream, topic) = shadow.pick_topic_pair(prng)?; + let live = *shadow + .topic_partitions + .get(&(stream.clone(), topic.clone()))?; + // One past the live count is the smallest guaranteed over-delete. + // Past the per-request cap the pre-consensus gate would answer + // `TooManyPartitions` instead, so the target is unrealizable. + let partitions_count = live.checked_add(1)?; + if partitions_count > MAX_PARTITIONS_PER_REQUEST { + return None; + } + Some(Input { + stream, + topic, + partitions_count, + }) + } } } @@ -96,7 +130,13 @@ pub const fn classify_reply(code: u32) -> Outcome { } #[must_use] -pub const fn predicted_effect(_input: &Input, _outcome: Outcome) -> Effect { - // Partition counts are not tracked per topic in the shadow. - Effect::None +pub fn predicted_effect(input: &Input, outcome: Outcome) -> Effect { + match outcome { + Outcome::Ok => Effect::RemovePartitions { + stream: input.stream.clone(), + topic: input.topic.clone(), + count: input.partitions_count, + }, + _ => Effect::None, + } } diff --git a/core/simulator/src/workload/shadow.rs b/core/simulator/src/workload/shadow.rs index 3d02b693e..69f64b2e4 100644 --- a/core/simulator/src/workload/shadow.rs +++ b/core/simulator/src/workload/shadow.rs @@ -46,6 +46,11 @@ pub struct Shadow { pub stream_names: IndexSet<String>, /// Live topics by `(stream, topic)`. Only added if parent stream lives. pub topic_names: IndexSet<(String, String)>, + /// Partition count per live topic, keyed like `topic_names`. Lets + /// `create/delete_partitions` sample in-bounds vs over-count deliberately + /// (the server rejects an over-count delete with a committed + /// `InvalidPartitionsCount`, so the count must be known at sample time). + pub topic_partitions: HashMap<(String, String), u32>, pub user_names: IndexSet<String>, pub pat_names: IndexSet<String>, pub consumer_group_names: IndexSet<(String, String, String)>, @@ -80,6 +85,7 @@ impl Shadow { namespaces_live, stream_names: IndexSet::new(), topic_names: IndexSet::new(), + topic_partitions: HashMap::new(), user_names: IndexSet::new(), pat_names: IndexSet::new(), consumer_group_names: IndexSet::new(), @@ -191,31 +197,23 @@ impl Shadow { let applied = match e { Effect::None => true, Effect::AddStream { name } => self.stream_names.insert(name), - Effect::RemoveStream { name } => { - let removed = self.stream_names.shift_remove(&name); - self.topic_names.retain(|(s, _)| s != &name); - self.consumer_group_names.retain(|(s, _, _)| s != &name); - removed - } + Effect::RemoveStream { name } => self.remove_stream(&name), Effect::AddTopic { stream, name, - partitions: _, - } => { - if self.stream_names.contains(&stream) { - self.topic_names.insert((stream, name)) - } else { - false - } - } - Effect::RemoveTopic { stream, name } => { - let removed = self - .topic_names - .shift_remove(&(stream.clone(), name.clone())); - self.consumer_group_names - .retain(|(s, t, _)| !(s == &stream && t == &name)); - removed - } + partitions, + } => self.add_topic(stream, name, partitions), + Effect::RemoveTopic { stream, name } => self.remove_topic(&stream, &name), + Effect::AddPartitions { + stream, + topic, + count, + } => self.add_partitions(stream, topic, count), + Effect::RemovePartitions { + stream, + topic, + count, + } => self.remove_partitions(stream, topic, count), Effect::AddUser { name } => { // Matches `create_user::sample`'s pw-{name} baseline. let password = format!("pw-{name}"); @@ -286,6 +284,60 @@ impl Shadow { } } + fn remove_stream(&mut self, name: &str) -> bool { + let removed = self.stream_names.shift_remove(name); + self.topic_names.retain(|(s, _)| s != name); + self.topic_partitions.retain(|(s, _), _| s != name); + self.consumer_group_names.retain(|(s, _, _)| s != name); + removed + } + + fn add_topic(&mut self, stream: String, name: String, partitions: u32) -> bool { + if self.stream_names.contains(&stream) { + self.topic_partitions + .insert((stream.clone(), name.clone()), partitions); + self.topic_names.insert((stream, name)) + } else { + false + } + } + + fn remove_topic(&mut self, stream: &str, name: &str) -> bool { + let removed = self + .topic_names + .shift_remove(&(stream.to_string(), name.to_string())); + self.topic_partitions + .remove(&(stream.to_string(), name.to_string())); + self.consumer_group_names + .retain(|(s, t, _)| !(s == stream && t == name)); + removed + } + + fn add_partitions(&mut self, stream: String, topic: String, count: u32) -> bool { + self.topic_partitions + .get_mut(&(stream, topic)) + .is_some_and(|partitions| { + *partitions = partitions.saturating_add(count); + true + }) + } + + /// The committed delete was in bounds on the server; a shadow count below + /// it means a concurrent commit defeated the sample-time precondition, + /// same as the other `applied = false` paths. + fn remove_partitions(&mut self, stream: String, topic: String, count: u32) -> bool { + self.topic_partitions + .get_mut(&(stream, topic)) + .is_some_and(|partitions| { + if *partitions >= count { + *partitions -= count; + true + } else { + false + } + }) + } + fn rename_stream(&mut self, old: &str, new: &str) -> bool { if !self.stream_names.shift_remove(old) { return false; @@ -304,6 +356,16 @@ impl Shadow { } }) .collect(); + self.topic_partitions = std::mem::take(&mut self.topic_partitions) + .into_iter() + .map(|((s, t), partitions)| { + if s == old { + ((new_owned.clone(), t), partitions) + } else { + ((s, t), partitions) + } + }) + .collect(); self.consumer_group_names = std::mem::take(&mut self.consumer_group_names) .into_iter() .map(|(s, t, g)| { @@ -326,6 +388,13 @@ impl Shadow { } self.topic_names .insert((stream.to_string(), new.to_string())); + if let Some(partitions) = self + .topic_partitions + .remove(&(stream.to_string(), old.to_string())) + { + self.topic_partitions + .insert((stream.to_string(), new.to_string()), partitions); + } let new_owned = new.to_string(); self.consumer_group_names = std::mem::take(&mut self.consumer_group_names) .into_iter() diff --git a/foreign/cpp/Cargo.toml b/foreign/cpp/Cargo.toml index d1f5afc2b..86e6672a5 100644 --- a/foreign/cpp/Cargo.toml +++ b/foreign/cpp/Cargo.toml @@ -29,7 +29,7 @@ crate-type = ["staticlib"] [dependencies] bytes = "1.12.1" cxx = "1.0.198" -iggy = { path = "../../core/sdk" } +iggy = { path = "../../core/sdk", features = ["vsr"] } iggy_binary_protocol = { path = "../../core/binary_protocol" } iggy_common = { path = "../../core/common" } # Explicitly enable the runtime + I/O drivers required by `Runtime::enable_all()` in lib.rs. diff --git a/foreign/cpp/tests/e2e/client.cpp b/foreign/cpp/tests/e2e/client.cpp index 64405fe92..53c1a96a5 100644 --- a/foreign/cpp/tests/e2e/client.cpp +++ b/foreign/cpp/tests/e2e/client.cpp @@ -448,40 +448,32 @@ TEST_F(LowLevelE2E_Client, GetClientsReflectsSessionRemovalAfterDisconnect) { } TEST_F(LowLevelE2E_Client, GetClientsReflectsLoggedOutSessionAsUnauthenticated) { - RecordProperty("description", - "Keeps a logged out session visible in get_clients and get_client, but marks it unauthenticated."); + RecordProperty("description", "Drops a logged out session from get_clients and reports it missing in get_client."); iggy::ffi::Client *first_client = GetLoggedInClient(); iggy::ffi::Client *second_client = GetLoggedInClient(); iggy::ffi::ClientInfoDetails first_me{}; - iggy::ffi::ClientInfoDetails logged_out_client{}; - rust::Vec<iggy::ffi::ClientInfo> clients_after_logout; ASSERT_NO_THROW({ first_me = first_client->get_me(); }); + // The VSR server drops the client-table entry on logout (an unauthenticated + // session is not tracked), unlike the legacy server which kept it visible + // without a user id. ASSERT_NO_THROW(first_client->logout_user()); - ASSERT_NO_THROW({ - clients_after_logout = second_client->get_clients(); - logged_out_client = second_client->get_client(first_me.client_id); - }); - - bool found_first = false; - for (const auto &client : clients_after_logout) { - if (client.client_id != first_me.client_id) { - continue; + constexpr auto removal_timeout = std::chrono::seconds(5); + constexpr auto removal_poll_interval = std::chrono::milliseconds(10); + const auto deadline = std::chrono::steady_clock::now() + removal_timeout; + bool removed = false; + do { + const auto clients = second_client->get_clients(); + removed = std::none_of(clients.begin(), clients.end(), + [&first_me](const auto &client) { return client.client_id == first_me.client_id; }); + if (removed) { + break; } - - found_first = true; - EXPECT_FALSE(client.has_user_id); - EXPECT_EQ(static_cast<std::string>(client.address), static_cast<std::string>(first_me.address)); - EXPECT_EQ(static_cast<std::string>(client.transport), static_cast<std::string>(first_me.transport)); - break; - } - - EXPECT_TRUE(found_first); - EXPECT_EQ(logged_out_client.client_id, first_me.client_id); - EXPECT_FALSE(logged_out_client.has_user_id); - EXPECT_EQ(static_cast<std::string>(logged_out_client.address), static_cast<std::string>(first_me.address)); - EXPECT_EQ(static_cast<std::string>(logged_out_client.transport), static_cast<std::string>(first_me.transport)); + std::this_thread::sleep_for(removal_poll_interval); + } while (std::chrono::steady_clock::now() < deadline); + ASSERT_TRUE(removed); + ASSERT_THROW(second_client->get_client(first_me.client_id), std::exception); } TEST_F(LowLevelE2E_Client, LoginWithoutConnect) { @@ -567,9 +559,12 @@ TEST_F(LowLevelE2E_Client, GetStatsBeforeLoginThrows) { ASSERT_THROW(client->get_stats(), std::exception); } -TEST_F(LowLevelE2E_Client, FlushUnsavedBufferSucceedsForExistingPartition) { +// The VSR server has no unsaved-buffer primitive (writes are journaled at +// commit); FLUSH_UNSAVED_BUFFER denies typed with FeatureUnavailable even for +// resolvable targets. +TEST_F(LowLevelE2E_Client, FlushUnsavedBufferThrowsForExistingPartition) { RecordProperty("description", - "Creates a stream and topic, sends one message, and flushes the partition buffer successfully."); + "Rejects flush_unsaved_buffer with the feature-unavailable error for an existing partition."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); iggy::ffi::Client *client = GetLoggedInClient(); @@ -585,13 +580,14 @@ TEST_F(LowLevelE2E_Client, FlushUnsavedBufferSucceedsForExistingPartition) { ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages))); - ASSERT_NO_THROW( - client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true)); + ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true), + std::exception); } -TEST_F(LowLevelE2E_Client, FlushUnsavedBufferSucceedsForExistingEmptyPartition) { - RecordProperty("description", - "Succeeds when flush_unsaved_buffer is called for an existing partition with no unsaved messages."); +TEST_F(LowLevelE2E_Client, FlushUnsavedBufferThrowsForExistingEmptyPartition) { + RecordProperty( + "description", + "Rejects flush_unsaved_buffer with the feature-unavailable error for a partition with no unsaved messages."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); iggy::ffi::Client *client = GetLoggedInClient(); @@ -602,8 +598,8 @@ TEST_F(LowLevelE2E_Client, FlushUnsavedBufferSucceedsForExistingEmptyPartition) ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0, "server_default")); - ASSERT_NO_THROW( - client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true)); + ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true), + std::exception); } TEST_F(LowLevelE2E_Client, FlushUnsavedBufferBeforeLoginThrows) { @@ -694,8 +690,9 @@ TEST_F(LowLevelE2E_Client, FlushUnsavedBufferAfterTopicDeletedThrows) { std::exception); } -TEST_F(LowLevelE2E_Client, FlushUnsavedBufferTwiceSucceeds) { - RecordProperty("description", "Allows flush_unsaved_buffer to be called twice in a row for the same partition."); +TEST_F(LowLevelE2E_Client, FlushUnsavedBufferTwiceThrows) { + RecordProperty("description", + "Rejects flush_unsaved_buffer with the feature-unavailable error consistently across repeat calls."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); iggy::ffi::Client *client = GetLoggedInClient(); @@ -711,10 +708,10 @@ TEST_F(LowLevelE2E_Client, FlushUnsavedBufferTwiceSucceeds) { ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages))); - ASSERT_NO_THROW( - client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true)); - ASSERT_NO_THROW( - client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true)); + ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true), + std::exception); + ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true), + std::exception); } TEST_F(LowLevelE2E_Client, FlushUnsavedBufferWithInvalidPartitionIdsThrows) { @@ -1566,15 +1563,19 @@ TEST_F(LowLevelE2E_Client, GetClientsReflectsAdditionalSession) { EXPECT_TRUE(found_after); } -TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginThrows) { +TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginSucceeds) { RecordProperty( "description", - "Rejects get_cluster_metadata before connect, after connect but before login, and after disconnect."); + "Serves get_cluster_metadata to a connected but unauthenticated client, and rejects it without a connection."); iggy::ffi::Client *client = GetLoggedOutClient(); ASSERT_THROW(client->get_cluster_metadata(), std::exception); ASSERT_NO_THROW(client->connect()); - ASSERT_THROW(client->get_cluster_metadata(), std::exception); + // By design pre-login on the VSR server: an SDK must read the roster to + // find the primary before it can authenticate (redirect bootstrap). + iggy::ffi::ClusterMetadata metadata{}; + ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); }); + ASSERT_EQ(metadata.nodes.size(), 1u); ASSERT_NO_THROW(client->login_user("iggy", "iggy")); ASSERT_NO_THROW(client->disconnect()); ASSERT_THROW(client->get_cluster_metadata(), std::exception); @@ -1631,6 +1632,8 @@ TEST_F(LowLevelE2E_Client, PingSucceedsForNewConnection) { RecordProperty("description", "Successfully pings the server from a fresh unauthenticated client session."); iggy::ffi::Client *client = GetLoggedOutClient(); + // The VSR client has no lazy connect; ping still needs no authentication. + ASSERT_NO_THROW(client->connect()); ASSERT_NO_THROW(client->ping()); } diff --git a/foreign/cpp/tests/e2e/consumer_group.cpp b/foreign/cpp/tests/e2e/consumer_group.cpp index 3ca41d3e1..7863b6630 100644 --- a/foreign/cpp/tests/e2e/consumer_group.cpp +++ b/foreign/cpp/tests/e2e/consumer_group.cpp @@ -722,19 +722,21 @@ TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReflectsJoinedGroupMembersCou EXPECT_NE(groups[0].members_count, groups[1].members_count); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentStreamReturnsEmpty) { - RecordProperty("description", "Returns an empty list when the stream does not exist."); +// The VSR server rejects consumer-group reads whose parent stream or topic is +// absent with the legacy typed not-found; the legacy server answered them with +// an empty list. +TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentStreamThrows) { + RecordProperty("description", "Throws when the stream does not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); iggy::ffi::Client *client = GetLoggedInClient(); - const auto groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); - EXPECT_TRUE(groups.empty()); + ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), + std::exception); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentTopicReturnsEmpty) { - RecordProperty("description", "Returns an empty list when the topic does not exist."); +TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentTopicThrows) { + RecordProperty("description", "Throws when the topic does not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); iggy::ffi::Client *client = GetLoggedInClient(); @@ -742,9 +744,8 @@ TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentTopicReturnsEmpt ASSERT_NO_THROW(client->create_stream(stream_name)); TrackStream(stream_name); - const auto groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); - EXPECT_TRUE(groups.empty()); + ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), + std::exception); } TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsIsStableAcrossBackToBackCalls) { @@ -836,8 +837,8 @@ TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReturnsCorrectNumberOfGroups) EXPECT_TRUE(groups_after_delete.empty()); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterStreamDeletionReturnsEmpty) { - RecordProperty("description", "Returns an empty list after deleting the stream that owned the groups."); +TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterStreamDeletionThrows) { + RecordProperty("description", "Throws after deleting the stream that owned the groups."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string first_group_name = GetRandomName(); @@ -859,13 +860,12 @@ TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterStreamDeletionReturnsEmp ForgetTrackedConsumerGroup(stream_name, topic_name, second_group_name); ForgetTrackedStream(stream_name); - const auto groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); - EXPECT_TRUE(groups.empty()); + ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), + std::exception); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterTopicDeletionReturnsEmpty) { - RecordProperty("description", "Returns an empty list after deleting the topic that owned the groups."); +TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterTopicDeletionThrows) { + RecordProperty("description", "Throws after deleting the topic that owned the groups."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string first_group_name = GetRandomName(); @@ -887,9 +887,8 @@ TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterTopicDeletionReturnsEmpt ForgetTrackedConsumerGroup(stream_name, topic_name, first_group_name); ForgetTrackedConsumerGroup(stream_name, topic_name, second_group_name); - const auto groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); - EXPECT_TRUE(groups.empty()); + ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), + std::exception); } TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupBeforeLoginThrows) { @@ -1134,7 +1133,10 @@ TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupAndRecreateWithSameNameSucc const auto recreated_group = client->create_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), group_name); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_EQ(recreated_group.id, 0u); + // The VSR server mints group ids monotonically; a recreate gets a + // fresh id (the deleted group held 0), unlike the legacy server which + // reused the freed slot. + ASSERT_GT(recreated_group.id, 0u); ASSERT_EQ(recreated_group.name, group_name); ASSERT_EQ(recreated_group.members_count, 0u); ASSERT_TRUE(recreated_group.members.empty()); diff --git a/foreign/cpp/tests/e2e/message.cpp b/foreign/cpp/tests/e2e/message.cpp index e20ef6008..f805127a3 100644 --- a/foreign/cpp/tests/e2e/message.cpp +++ b/foreign/cpp/tests/e2e/message.cpp @@ -51,9 +51,10 @@ TEST_F(LowLevelE2E_Message, SendAndPollMessagesRoundTrip) { ASSERT_NO_THROW(sent = client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages))); - ASSERT_TRUE(sent.confirmations.empty()) - << "The legacy server reports no offsets, so the confirmation list must stay empty, got " - << sent.confirmations.size(); + ASSERT_EQ(sent.confirmations.size(), 1u) + << "The VSR server reports the written partition's offsets, so a single-partition send " + << "must carry exactly one confirmation"; + EXPECT_EQ(sent.confirmations.front().partition_id, 0u); auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", make_numeric_identifier(1), "offset", 0, 100, false); diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py index 417f24a38..46434525e 100644 --- a/foreign/python/tests/test_topic.py +++ b/foreign/python/tests/test_topic.py @@ -24,7 +24,6 @@ from apache_iggy import IggyClient, IggyExpiry, MaxTopicSize, SendMessage from .utils import ( get_server_config, wait_for_ping, - wait_for_purged_topic, wait_for_server, ) @@ -1384,9 +1383,10 @@ class TestPurgeTopic: await iggy_client.purge_topic(stream_name, topic_name) - # The purge ack precedes the asynchronous partition prune, so poll - # until the stats catch up instead of asserting the counts directly. - after = await wait_for_purged_topic(iggy_client, stream_name, topic_name) + after = await iggy_client.get_topic(stream_name, topic_name) + assert after is not None + assert after.messages_count == 0 + assert after.size == 0 # Purging clears messages and size only; topic config is unchanged. assert after.id == before.id assert after.name == before.name @@ -1437,7 +1437,10 @@ class TestPurgeTopic: await iggy_client.purge_topic(stream_name, topic_name) await iggy_client.purge_topic(stream_name, topic_name) - await wait_for_purged_topic(iggy_client, stream_name, topic_name) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.messages_count == 0 + assert topic.size == 0 @pytest.mark.asyncio async def test_purge_nonexistent_topic_fails( diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py index b3ea85c41..b37a53831 100644 --- a/foreign/python/tests/utils.py +++ b/foreign/python/tests/utils.py @@ -24,7 +24,7 @@ import os import socket import time -from apache_iggy import IggyClient, TopicDetails +from apache_iggy import IggyClient # Server-side limits: usernames are 3-50 bytes, passwords 3-100 bytes. MIN_USERNAME_BYTES = 3 @@ -115,37 +115,6 @@ async def wait_for_ping( await asyncio.sleep(interval) -async def wait_for_purged_topic( - client: IggyClient, stream: str, topic: str, timeout: float = 10.0 -) -> TopicDetails: - """ - Poll get_topic until a committed purge is reflected in the stats. - - The VSR server acknowledges a purge once it commits; partition data - is pruned asynchronously, so stats can transiently report pre-purge - counts. - - Returns: - TopicDetails once messages_count and size reach 0 - - Raises: - TimeoutError: If the purge is not reflected within timeout - """ - deadline = time.time() + timeout - - while True: - details = await client.get_topic(stream, topic) - assert details is not None, "purged topic must still exist" - if details.messages_count == 0 and details.size == 0: - return details - if time.time() >= deadline: - raise TimeoutError( - f"purge of {stream}/{topic} not reflected after {timeout}s: " - f"messages_count={details.messages_count} size={details.size}" - ) - await asyncio.sleep(0.05) - - def unique_credentials(unique_name) -> tuple[str, str]: """Return a unique (username, password) pair within the server limits.""" username = unique_name(max_bytes=MAX_USERNAME_BYTES) diff --git a/scripts/run-bdd-tests.sh b/scripts/run-bdd-tests.sh index c95de1cb4..f23b1de9b 100755 --- a/scripts/run-bdd-tests.sh +++ b/scripts/run-bdd-tests.sh @@ -40,7 +40,7 @@ usage(){ log " sdk: rust | python | php | go | go-race | node | csharp | java | cpp | all | clean (default: all)" log " feature: basic_messaging | leader_redirection | raw_command | all (default: all)" # TODO: change to iggy-server once legacy server is removed (core/server has VSR support) - log " --vsr: run against iggy-server-ng built with --features vsr (rust, python, go, node, csharp);" + log " --vsr: run against iggy-server-ng built with --features vsr (rust, python, go, node, csharp, cpp);" log " expects IGGY_SERVER_NG_PATH (default: target/debug/iggy-server-ng)" log " and a vsr-built iggy CLI at IGGY_CLI_PATH." log " The go suites imply it: the Go SDK speaks only the VSR protocol." @@ -56,12 +56,12 @@ usage(){ if [ "$VSR" = "1" ]; then case "$SDK" in - rust|python|go|go-race|node|csharp|clean) ;; + rust|python|go|go-race|node|csharp|cpp|clean) ;; java) # Redundant: the Java suite applies the VSR overlay unconditionally. VSR=0 ;; *) - log "❌ --vsr supports only the Rust, Python, Go, Node, C#, and Java SDKs so far" + log "❌ --vsr supports only the Rust, Python, Go, Node, C#, Java, and C++ SDKs so far" usage exit 2 ;; esac
