You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/08/10 14:11:39 UTC
(iggy) branch master updated: feat(cpp): run C++ SDK tests only against the VSR server (#3844)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 00840d188 feat(cpp): run C++ SDK tests only against the VSR server (#3844)
00840d188 is described below
commit 00840d188b96a53c36e2e7b9cfbd352c5d01fb57
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Aug 10 16:11:35 2026 +0200
feat(cpp): run C++ SDK tests only against the VSR server (#3844)
---
.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 +-
.../FetchMessagesTests.cs | 29 +-
foreign/python/tests/test_topic.py | 13 +-
foreign/python/tests/utils.py | 33 +-
scripts/run-bdd-tests.sh | 6 +-
24 files changed, 962 insertions(+), 250 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 e391a5679..ce6faab07 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;
@@ -1544,6 +1545,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,
@@ -1551,6 +1559,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);
@@ -1943,14 +1952,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/csharp/Iggy_SDK.Tests.Integration/FetchMessagesTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/FetchMessagesTests.cs
index c51e0c052..f9ae474a2 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/FetchMessagesTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/FetchMessagesTests.cs
@@ -91,37 +91,12 @@ public class FetchMessagesTests
[Test]
[MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
- [SkipHttp]
- public async Task PollMessages_InvalidTopic_Should_ReturnEmptyPoll(Protocol protocol)
- {
- var (client, streamName) = await CreateStreamWithMessages(protocol);
-
- // TCP replies status 0 with an empty poll (partition 0, offset 0) so the
- // SDK fails fast on decode instead of hanging until its read timeout.
- var response = await client.PollMessagesAsync(new MessageFetchRequest
- {
- Count = 10,
- AutoCommit = true,
- Consumer = Consumer.New(1),
- PartitionId = 0,
- PollingStrategy = PollingStrategy.Next(),
- StreamId = Identifier.String(streamName),
- TopicId = Identifier.Numeric(2137)
- });
-
- response.Messages.ShouldBeEmpty();
- response.PartitionId.ShouldBe(0);
- response.CurrentOffset.ShouldBe(0u);
- }
-
- [Test]
- [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
- [SkipTcp]
public async Task PollMessages_InvalidTopic_Should_Throw_NotFound(Protocol protocol)
{
var (client, streamName) = await CreateStreamWithMessages(protocol);
- // HTTP keeps legacy-server parity and answers 404.
+ // A missing topic is an addressing error, not an empty partition:
+ // HTTP answers 404, TCP the typed topic-not-found rejection.
await Should.ThrowAsync<IggyInvalidStatusCodeException>(() =>
client.PollMessagesAsync(new MessageFetchRequest
{
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