test(integration): remove suites duplicating existing coverage (#3823)
diff --git a/Cargo.lock b/Cargo.lock
index 3cd7dcc..b2af9ce 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11933,7 +11933,6 @@
"hwlocality",
"hyper",
"hyper-util",
- "iggy",
"iggy_binary_protocol",
"iggy_common",
"journal",
diff --git a/core/integration/tests/cluster/metadata_checkpoint_restart.rs b/core/integration/tests/cluster/metadata_checkpoint_restart.rs
index 79f3f82..75ae60f 100644
--- a/core/integration/tests/cluster/metadata_checkpoint_restart.rs
+++ b/core/integration/tests/cluster/metadata_checkpoint_restart.rs
@@ -16,7 +16,8 @@
// under the License.
//! Checkpoint-shaped edge cases for metadata + client-table state transfer,
-//! metadata-plane commands only (`CreateStream` over the raw VSR wire).
+//! metadata-plane commands only (`CreateStream`). The 3-node cases drive the
+//! raw VSR wire; the solo case at the bottom uses the regular TCP client.
//!
//! `metadata_state_transfer` pins the base case: one checkpoint, a live
//! post-checkpoint tail, one restart. This file covers the shapes around it:
@@ -30,22 +31,27 @@
//! from the TRANSFERRED reply ring - the WAL entry that produced the reply
//! was drained on every node, so nothing but the shipped table can answer;
//! - a second restart of the same node (the installed snapshot must persist
-//! and serve as the local floor for another transfer round).
+//! and serve as the local floor for another transfer round);
+//! - a SOLO replica restarting after its own checkpoint: no transfer in the
+//! picture, so recovery has to fold the local snapshot back in as the floor
+//! and replay the WAL suffix on top.
//!
-//! Checkpoint placement is config-driven: with `metadata.journal_slots = 256`
-//! and the built-in checkpoint margin (64), a checkpoint fires when the
-//! journal holds 192 committed ops, i.e. at op 192, 384, ... The register
-//! commits at op 1 and request K at op K + 1, so request 191 lands checkpoint
-//! one and request 383 lands checkpoint two. Requests here are sequential
-//! with one in flight, which keeps that arithmetic exact; the
-//! `forced checkpoint completed` markers below pin it rather than trust it.
+//! Checkpoint placement in the 3-node cases is config-driven: with
+//! `metadata.journal_slots = 256` and the built-in checkpoint margin (64), a
+//! checkpoint fires when the journal holds 192 committed ops, i.e. at op 192,
+//! 384, ... The register commits at op 1 and request K at op K + 1, so request
+//! 191 lands checkpoint one and request 383 lands checkpoint two. Requests
+//! here are sequential with one in flight, which keeps that arithmetic exact;
+//! the `forced checkpoint completed` markers below pin it rather than trust
+//! it.
#![cfg(feature = "vsr")]
use super::client_table_restart::{
commit_request, create_stream_payload, register, resume_request, tcp_addr, tcp_addrs,
};
-use integration::harness::TestHarness;
+use iggy::prelude::*;
+use integration::harness::{TestBinary, TestHarness};
use integration::iggy_harness;
use std::time::Duration;
use tokio::time::{Instant, sleep};
@@ -274,3 +280,115 @@
)
.await;
}
+
+/// The prepare WAL has `SLOT_COUNT = 1024` slots and the coordinator checkpoints at
+/// `<= CHECKPOINT_MARGIN` (64) remaining, so after ~960 uncheckpointed ops. 1024
+/// stream creates clears that with room for a WAL suffix above the checkpoint, and
+/// stays under the 4096 stream namespace cap.
+const STREAMS: u32 = 1024;
+
+const RECOVER_TIMEOUT: Duration = Duration::from_secs(60);
+const POLL_INTERVAL: Duration = Duration::from_millis(200);
+
+fn stream_name(index: u32) -> String {
+ format!("ckpt-stream-{index}")
+}
+
+/// Connect a root-authenticated TCP client to the solo node.
+async fn connect(harness: &TestHarness) -> IggyClient {
+ harness
+ .node(0)
+ .tcp_client()
+ .expect("tcp client builder")
+ .with_root_login()
+ .connect()
+ .await
+ .expect("connect to the solo node")
+}
+
+/// Poll the solo node until it is back up and serving `stream`, returning the connected
+/// client. Panics on timeout.
+async fn wait_for_stream(harness: &TestHarness, stream: &str) -> IggyClient {
+ let stream_id = Identifier::named(stream).unwrap();
+ let deadline = Instant::now() + RECOVER_TIMEOUT;
+ loop {
+ if let Ok(builder) = harness.node(0).tcp_client()
+ && let Ok(client) = builder.with_root_login().connect().await
+ && matches!(client.get_stream(&stream_id).await, Ok(Some(_)))
+ {
+ return client;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "solo node did not recover and serve {stream} within {RECOVER_TIMEOUT:?}"
+ );
+ sleep(POLL_INTERVAL).await;
+ }
+}
+
+// Metadata checkpoint-fold recovery across a solo restart, over
+// `iggy-server-ng`'s production snapshot and WAL path.
+//
+// Between checkpoints a replica recovers its metadata by replaying the WAL. Once the
+// WAL fills, the `SnapshotCoordinator` checkpoints: it persists `snapshot.bin`, pairs
+// it in the superblock, and DRAINS the snapshotted prefix from the WAL. A restart
+// after that must fold the snapshot back in as the recovery floor and replay only the
+// committed suffix on top, rather than rely on a full WAL replay, which no longer
+// holds the drained ops. This drives a real checkpoint by pushing the metadata WAL
+// past `CHECKPOINT_MARGIN`, restarts the process, and asserts that a stream from the
+// drained prefix, recoverable only from the snapshot, and one from the WAL suffix
+// both survive.
+//
+// Solo on purpose: 1-of-1 quorum commits every op the instant it is journaled, so
+// bulk creation is fast and the WAL is fully committed with no uncommitted suffix to
+// reconcile, exercising checkpoint and snapshot-fold recovery in isolation without an
+// election in the mix.
+#[iggy_harness(cluster_nodes = 1, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_checkpointed_metadata_when_solo_replica_restarts_should_recover_from_snapshot_and_wal(
+ harness: &mut TestHarness,
+) {
+ let client = connect(harness).await;
+ for index in 0..STREAMS {
+ client
+ .create_stream(&stream_name(index))
+ .await
+ .unwrap_or_else(|e| panic!("create stream {index}: {e}"));
+ }
+ drop(client);
+
+ // Crossing CHECKPOINT_MARGIN must have driven the coordinator to persist a snapshot
+ // and drain the WAL prefix behind it.
+ let snapshot_path = harness
+ .node(0)
+ .data_path()
+ .join("metadata")
+ .join("snapshot.bin");
+ let snapshot_len = std::fs::metadata(&snapshot_path).map(|m| m.len()).ok();
+ assert!(
+ snapshot_len.is_some_and(|len| len > 0),
+ "{STREAMS} committed metadata ops must cross CHECKPOINT_MARGIN and persist a \
+ non-empty snapshot at {}, got {snapshot_len:?}",
+ snapshot_path.display()
+ );
+
+ // Restart the solo node: recovery loads the snapshot, holding the drained prefix,
+ // and replays the committed WAL suffix on top.
+ harness.node_mut(0).stop().expect("stop the solo node");
+ harness.node_mut(0).start().expect("restart the solo node");
+
+ // The first stream sits far below the checkpoint op, so it was drained from the WAL
+ // and can come back only from the snapshot; the last stream is in the WAL suffix
+ // above the checkpoint. Both surviving proves snapshot-fold plus suffix recovery,
+ // not a bare WAL replay.
+ let client = wait_for_stream(harness, &stream_name(0)).await;
+ for name in [stream_name(0), stream_name(STREAMS - 1)] {
+ assert!(
+ client
+ .get_stream(&Identifier::named(&name).unwrap())
+ .await
+ .expect("get stream after restart")
+ .is_some(),
+ "stream {name} must survive the checkpointed restart"
+ );
+ }
+}
diff --git a/core/integration/tests/sdk/mcp_parity.rs b/core/integration/tests/sdk/mcp_parity.rs
deleted file mode 100644
index 4ec5422..0000000
--- a/core/integration/tests/sdk/mcp_parity.rs
+++ /dev/null
@@ -1,608 +0,0 @@
-// 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.
-
-//! Plain-SDK reproduction of the `mcp::*` suite, without the MCP server. Each
-//! MCP tool maps 1:1 to an SDK client method, so these tests exercise the same
-//! server-ng behaviour under the `vsr` feature over TCP, without the MCP/HTTP
-//! environment overhead. Assertions mirror the MCP counterparts.
-
-#![cfg(feature = "vsr")]
-
-use iggy::prelude::*;
-use integration::{
- harness::{TestHarness, seeds},
- iggy_harness,
-};
-
-/// The consumer under which `seeds::mcp_standard` stores its offset (matches the
-/// MCP server's default consumer name).
-fn seed_consumer() -> Consumer {
- Consumer::new(Identifier::named(seeds::names::CONSUMER).unwrap())
-}
-
-async fn root_client(harness: &TestHarness) -> IggyClient {
- let client = harness.new_client().await.unwrap();
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
- client
-}
-
-// mcp::should_handle_ping
-#[iggy_harness(test_client_transport = [Tcp])]
-async fn should_handle_ping(harness: &TestHarness) {
- let client = root_client(harness).await;
- client.ping().await.unwrap();
-}
-
-// mcp::should_return_cluster_metadata
-// TODO: MCP polls `get_cluster_metadata` via a dedicated tool; there is no
-// equivalent single SDK client method to assert cluster name + node count, so
-// this case is intentionally not ported here.
-
-// mcp::should_return_list_of_streams
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_list_of_streams(harness: &TestHarness) {
- let client = root_client(harness).await;
- let streams = client.get_streams().await.unwrap();
-
- assert_eq!(streams.len(), 1);
- assert_eq!(streams[0].name, seeds::names::STREAM);
- assert_eq!(streams[0].topics_count, 1);
-}
-
-// mcp::should_return_stream_details
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_stream_details(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let stream = client
- .get_stream(&stream_id)
- .await
- .unwrap()
- .expect("stream exists");
-
- assert_eq!(stream.name, seeds::names::STREAM);
- assert_eq!(stream.topics_count, 1);
- assert_eq!(stream.messages_count, 1);
-}
-
-// mcp::should_create_stream
-#[iggy_harness(test_client_transport = [Tcp])]
-async fn should_create_stream(harness: &TestHarness) {
- let client = root_client(harness).await;
- let name = "new_stream";
- let stream = client.create_stream(name).await.unwrap();
-
- assert_eq!(stream.name, name);
- assert_eq!(stream.topics_count, 0);
- assert_eq!(stream.messages_count, 0);
-}
-
-// mcp::should_update_stream
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_update_stream(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- client
- .update_stream(&stream_id, "updated_stream")
- .await
- .unwrap();
-}
-
-// mcp::should_delete_stream
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_stream(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- client.delete_stream(&stream_id).await.unwrap();
-}
-
-// mcp::should_purge_stream
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_purge_stream(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- client.purge_stream(&stream_id).await.unwrap();
-}
-
-// mcp::should_return_list_of_topics
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_list_of_topics(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topics = client.get_topics(&stream_id).await.unwrap();
-
- assert_eq!(topics.len(), 1);
- assert_eq!(topics[0].name, seeds::names::TOPIC);
- assert_eq!(topics[0].partitions_count, 1);
- assert_eq!(topics[0].messages_count, 1);
-}
-
-// mcp::should_return_topic_details
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_topic_details(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let topic = client
- .get_topic(&stream_id, &topic_id)
- .await
- .unwrap()
- .expect("topic exists");
-
- assert_eq!(topic.id, 0);
- assert_eq!(topic.name, seeds::names::TOPIC);
- assert_eq!(topic.messages_count, 1);
-}
-
-// mcp::should_create_topic
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_create_topic(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let name = "new_topic";
- let topic = client
- .create_topic(
- &stream_id,
- name,
- 1,
- CompressionAlgorithm::None,
- None,
- IggyExpiry::ServerDefault,
- MaxTopicSize::ServerDefault,
- )
- .await
- .unwrap();
-
- assert_eq!(topic.id, 1);
- assert_eq!(topic.name, name);
- assert_eq!(topic.partitions_count, 1);
- assert_eq!(topic.messages_count, 0);
-}
-
-// mcp::should_update_topic
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_update_topic(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client
- .update_topic(
- &stream_id,
- &topic_id,
- "updated_topic",
- CompressionAlgorithm::None,
- None,
- IggyExpiry::ServerDefault,
- MaxTopicSize::ServerDefault,
- )
- .await
- .unwrap();
-}
-
-// mcp::should_delete_topic
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_topic(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client.delete_topic(&stream_id, &topic_id).await.unwrap();
-}
-
-// mcp::should_purge_topic
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_purge_topic(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client.purge_topic(&stream_id, &topic_id).await.unwrap();
-}
-
-// mcp::should_create_partitions
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_create_partitions(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client
- .create_partitions(&stream_id, &topic_id, 3)
- .await
- .unwrap();
-}
-
-// mcp::should_delete_partitions
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_partitions(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client
- .delete_partitions(&stream_id, &topic_id, 1)
- .await
- .unwrap();
-}
-
-// mcp::should_delete_segments
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_segments(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client
- .delete_segments(&stream_id, &topic_id, 0, 1)
- .await
- .unwrap();
-}
-
-// mcp::should_poll_messages
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_poll_messages(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let messages = client
- .poll_messages(
- &stream_id,
- &topic_id,
- Some(0),
- &seed_consumer(),
- &PollingStrategy::offset(0),
- 10,
- false,
- )
- .await
- .unwrap();
-
- assert_eq!(messages.messages.len(), 1);
- assert_eq!(messages.messages[0].header.offset, 0);
- let payload = messages.messages[0]
- .payload_as_string()
- .expect("Failed to parse payload");
- assert_eq!(payload, seeds::names::MESSAGE_PAYLOAD);
-}
-
-// mcp::should_send_messages
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_send_messages(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let mut messages = vec![
- IggyMessage::builder()
- .payload("test".into())
- .build()
- .unwrap(),
- ];
- client
- .send_messages(
- &stream_id,
- &topic_id,
- &Partitioning::partition_id(0),
- &mut messages,
- )
- .await
- .unwrap();
-}
-
-// mcp::should_return_stats
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_stats(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stats = client.get_stats().await.unwrap();
-
- assert!(!stats.hostname.is_empty());
- assert_eq!(stats.messages_count, 1);
-}
-
-// mcp::should_return_me
-#[iggy_harness(test_client_transport = [Tcp])]
-async fn should_return_me(harness: &TestHarness) {
- let client = root_client(harness).await;
- let me = client.get_me().await.unwrap();
-
- assert!(me.client_id > 0);
-}
-
-// mcp::should_return_clients
-#[iggy_harness(test_client_transport = [Tcp])]
-async fn should_return_clients(harness: &TestHarness) {
- let client = root_client(harness).await;
- let clients = client.get_clients().await.unwrap();
-
- assert!(!clients.is_empty());
-}
-
-// mcp::should_handle_snapshot
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_handle_snapshot(harness: &TestHarness) {
- let client = root_client(harness).await;
- let snapshot = client
- .snapshot(
- SnapshotCompression::default(),
- SystemSnapshotType::all_snapshot_types(),
- )
- .await
- .unwrap();
-
- assert!(!snapshot.0.is_empty());
-}
-
-// mcp::should_return_consumer_groups
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_consumer_groups(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let groups = client
- .get_consumer_groups(&stream_id, &topic_id)
- .await
- .unwrap();
-
- assert!(!groups.is_empty());
-}
-
-// mcp::should_return_consumer_group_details
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_consumer_group_details(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let group_id = Identifier::named(seeds::names::CONSUMER_GROUP).unwrap();
- let group = client
- .get_consumer_group(&stream_id, &topic_id, &group_id)
- .await
- .unwrap()
- .expect("consumer group exists");
-
- assert_eq!(group.name, seeds::names::CONSUMER_GROUP);
- assert_eq!(group.partitions_count, 1);
- assert_eq!(group.members_count, 0);
- assert!(group.members.is_empty());
-}
-
-// mcp::should_create_consumer_group
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_create_consumer_group(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let name = "new_group";
- let group = client
- .create_consumer_group(&stream_id, &topic_id, name)
- .await
- .unwrap();
-
- assert_eq!(group.name, name);
- assert_eq!(group.partitions_count, 1);
- assert_eq!(group.members_count, 0);
- assert!(group.members.is_empty());
-}
-
-// mcp::should_delete_consumer_group
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_consumer_group(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let group_id = Identifier::named(seeds::names::CONSUMER_GROUP).unwrap();
- client
- .delete_consumer_group(&stream_id, &topic_id, &group_id)
- .await
- .unwrap();
-}
-
-// mcp::should_return_consumer_offset
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_consumer_offset(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let offset = client
- .get_consumer_offset(&seed_consumer(), &stream_id, &topic_id, Some(0))
- .await
- .unwrap()
- .expect("Expected consumer offset");
-
- assert_eq!(offset.partition_id, 0);
- assert_eq!(offset.stored_offset, 0);
- assert_eq!(offset.current_offset, 0);
-}
-
-// mcp::should_store_consumer_offset
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_store_consumer_offset(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client
- .store_consumer_offset(&seed_consumer(), &stream_id, &topic_id, Some(0), 0)
- .await
- .unwrap();
-}
-
-// mcp::should_delete_consumer_offset
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_consumer_offset(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- client
- .delete_consumer_offset(&seed_consumer(), &stream_id, &topic_id, Some(0))
- .await
- .unwrap();
-}
-
-// Regression (no MCP counterpart): deleting an offset that was never stored
-// must return a terminal `ConsumerOffsetNotFound`, NOT hang. The primary rejects
-// at admission with an explicit error reply instead of dropping the request
-// (which left the SDK replaying until its read-timeout).
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn delete_of_missing_consumer_offset_returns_not_found(harness: &TestHarness) {
- let client = root_client(harness).await;
- let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
- let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
- let never_stored = Consumer::new(Identifier::named("never_stored").unwrap());
- let result = client
- .delete_consumer_offset(&never_stored, &stream_id, &topic_id, Some(0))
- .await;
- assert!(
- matches!(result, Err(IggyError::ConsumerOffsetNotFound(_))),
- "expected ConsumerOffsetNotFound, got {result:?}"
- );
-}
-
-// mcp::should_return_personal_access_tokens
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_personal_access_tokens(harness: &TestHarness) {
- let client = root_client(harness).await;
- let tokens = client.get_personal_access_tokens().await.unwrap();
-
- assert_eq!(tokens.len(), 1);
- assert_eq!(tokens[0].name, seeds::names::PERSONAL_ACCESS_TOKEN);
-}
-
-// mcp::should_create_personal_access_token
-#[iggy_harness(test_client_transport = [Tcp])]
-async fn should_create_personal_access_token(harness: &TestHarness) {
- let client = root_client(harness).await;
- let name = "test_token";
- let token = client
- .create_personal_access_token(name, PersonalAccessTokenExpiry::NeverExpire)
- .await
- .unwrap();
-
- assert!(!token.token.is_empty());
-}
-
-// mcp::should_delete_personal_access_token
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_personal_access_token(harness: &TestHarness) {
- let client = root_client(harness).await;
- client
- .delete_personal_access_token(seeds::names::PERSONAL_ACCESS_TOKEN)
- .await
- .unwrap();
-}
-
-// mcp::should_return_users
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_users(harness: &TestHarness) {
- let client = root_client(harness).await;
- let users = client.get_users().await.unwrap();
-
- assert_eq!(users.len(), 2);
-}
-
-// mcp::should_return_user_details
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_return_user_details(harness: &TestHarness) {
- let client = root_client(harness).await;
- let user_id = Identifier::named(seeds::names::USER).unwrap();
- let user = client
- .get_user(&user_id)
- .await
- .unwrap()
- .expect("user exists");
-
- assert_eq!(user.username, seeds::names::USER);
-}
-
-// mcp::should_create_user
-#[iggy_harness(test_client_transport = [Tcp])]
-async fn should_create_user(harness: &TestHarness) {
- let client = root_client(harness).await;
- let username = "test-mcp-user";
- let user = client
- .create_user(username, "secret", UserStatus::Active, None)
- .await
- .unwrap();
-
- assert_eq!(user.username, username);
-}
-
-// mcp::should_update_user
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_update_user(harness: &TestHarness) {
- let client = root_client(harness).await;
- let user_id = Identifier::named(seeds::names::USER).unwrap();
- client
- .update_user(&user_id, Some("updated-user"), Some(UserStatus::Inactive))
- .await
- .unwrap();
-}
-
-// mcp::should_delete_user
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_delete_user(harness: &TestHarness) {
- let client = root_client(harness).await;
- let user_id = Identifier::named(seeds::names::USER).unwrap();
- client.delete_user(&user_id).await.unwrap();
-}
-
-// mcp::should_update_permissions
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_update_permissions(harness: &TestHarness) {
- let client = root_client(harness).await;
- let user_id = Identifier::named(seeds::names::USER).unwrap();
-
- let mut topics = std::collections::BTreeMap::new();
- topics.insert(
- 1,
- TopicPermissions {
- manage_topic: true,
- read_topic: true,
- poll_messages: true,
- send_messages: true,
- },
- );
- let mut streams = std::collections::BTreeMap::new();
- streams.insert(
- 1,
- StreamPermissions {
- manage_stream: true,
- manage_topics: true,
- topics: Some(topics),
- ..Default::default()
- },
- );
- let permissions = Permissions {
- global: GlobalPermissions {
- manage_servers: true,
- read_users: true,
- ..Default::default()
- },
- streams: Some(streams),
- };
-
- client
- .update_permissions(&user_id, Some(permissions))
- .await
- .unwrap();
-}
-
-// mcp::should_change_password
-#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
-async fn should_change_password(harness: &TestHarness) {
- let client = root_client(harness).await;
- let user_id = Identifier::named(seeds::names::USER).unwrap();
- client
- .change_password(&user_id, seeds::names::USER_PASSWORD, "new_secret")
- .await
- .unwrap();
-}
diff --git a/core/integration/tests/sdk/messages.rs b/core/integration/tests/sdk/messages.rs
deleted file mode 100644
index 34ce3d7..0000000
--- a/core/integration/tests/sdk/messages.rs
+++ /dev/null
@@ -1,172 +0,0 @@
-// 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.
-
-//! VSR data-plane round trip: `send_messages` / `poll_messages` and the
-//! consumer-offset store/get pair against a 3-node cluster, exercising the
-//! partition-reconciliation loop end to end.
-
-#![cfg(feature = "vsr")]
-
-use iggy::prelude::*;
-use integration::iggy_harness;
-
-// server-ng partition ids are 0-based (CreateTopic assigns them from 0).
-const PARTITION_ID: u32 = 0;
-const MESSAGES_COUNT: u32 = 10;
-
-#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
-async fn send_and_poll_messages_round_trip(harness: &TestHarness) {
- let client = harness.new_client().await.unwrap();
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
-
- client.create_stream("data-stream").await.unwrap();
- client
- .create_topic(
- &Identifier::named("data-stream").unwrap(),
- "data-topic",
- 1,
- CompressionAlgorithm::default(),
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .unwrap();
-
- let mut messages: Vec<IggyMessage> = (0..MESSAGES_COUNT)
- .map(|i| {
- IggyMessage::builder()
- .id(u128::from(i + 1))
- .payload(format!("payload-{i}").into())
- .build()
- .expect("message build")
- })
- .collect();
-
- client
- .send_messages(
- &Identifier::named("data-stream").unwrap(),
- &Identifier::named("data-topic").unwrap(),
- &Partitioning::partition_id(PARTITION_ID),
- &mut messages,
- )
- .await
- .expect("send_messages");
-
- let consumer = Consumer::default();
- let polled = client
- .poll_messages(
- &Identifier::named("data-stream").unwrap(),
- &Identifier::named("data-topic").unwrap(),
- Some(PARTITION_ID),
- &consumer,
- &PollingStrategy::offset(0),
- MESSAGES_COUNT,
- false,
- )
- .await
- .expect("poll_messages");
-
- assert_eq!(
- polled.messages.len() as u32,
- MESSAGES_COUNT,
- "all sent messages must come back"
- );
- for (i, message) in polled.messages.iter().enumerate() {
- assert_eq!(message.header.offset, i as u64, "offsets must be dense");
- assert_eq!(
- message.payload,
- bytes::Bytes::from(format!("payload-{i}")),
- "payload round trip for message {i}"
- );
- }
-
- client.logout_user().await.unwrap();
-}
-
-#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
-async fn consumer_offset_store_get_round_trip(harness: &TestHarness) {
- let client = harness.new_client().await.unwrap();
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
-
- client.create_stream("offset-stream").await.unwrap();
- client
- .create_topic(
- &Identifier::named("offset-stream").unwrap(),
- "offset-topic",
- 1,
- CompressionAlgorithm::default(),
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .unwrap();
-
- let mut messages: Vec<IggyMessage> = (0..5u32)
- .map(|i| {
- IggyMessage::builder()
- .id(u128::from(i + 1))
- .payload(format!("offset-payload-{i}").into())
- .build()
- .expect("message build")
- })
- .collect();
- client
- .send_messages(
- &Identifier::named("offset-stream").unwrap(),
- &Identifier::named("offset-topic").unwrap(),
- &Partitioning::partition_id(PARTITION_ID),
- &mut messages,
- )
- .await
- .expect("send_messages");
-
- let consumer = Consumer::default();
- client
- .store_consumer_offset(
- &consumer,
- &Identifier::named("offset-stream").unwrap(),
- &Identifier::named("offset-topic").unwrap(),
- Some(PARTITION_ID),
- 3,
- )
- .await
- .expect("store_consumer_offset");
-
- let stored = client
- .get_consumer_offset(
- &consumer,
- &Identifier::named("offset-stream").unwrap(),
- &Identifier::named("offset-topic").unwrap(),
- Some(PARTITION_ID),
- )
- .await
- .expect("get_consumer_offset")
- .expect("offset must exist after store");
-
- assert_eq!(stored.stored_offset, 3, "stored offset must round trip");
- assert_eq!(stored.partition_id, PARTITION_ID);
-
- client.logout_user().await.unwrap();
-}
diff --git a/core/integration/tests/sdk/mod.rs b/core/integration/tests/sdk/mod.rs
index 7d72480..1182d9e 100644
--- a/core/integration/tests/sdk/mod.rs
+++ b/core/integration/tests/sdk/mod.rs
@@ -21,10 +21,6 @@
mod hello_world;
#[cfg(feature = "vsr")]
mod http_refresh;
-#[cfg(feature = "vsr")]
-mod mcp_parity;
-#[cfg(feature = "vsr")]
-mod messages;
mod producer;
#[cfg(feature = "vsr")]
mod protocol_version;
diff --git a/core/integration/tests/server/general.rs b/core/integration/tests/server/general.rs
index cb51ac3..c9a5ad6 100644
--- a/core/integration/tests/server/general.rs
+++ b/core/integration/tests/server/general.rs
@@ -18,9 +18,9 @@
#[cfg(not(feature = "vsr"))]
use crate::server::scenarios::bench_scenario;
use crate::server::scenarios::{
- authentication_scenario, consumer_timestamp_polling_scenario, create_message_payload,
- invalid_consumer_offset_scenario, message_headers_scenario, permissions_scenario,
- snapshot_scenario, stream_size_validation_scenario, system_scenario, user_scenario,
+ authentication_scenario, consumer_timestamp_polling_scenario, invalid_consumer_offset_scenario,
+ message_headers_scenario, permissions_scenario, snapshot_scenario,
+ stream_size_validation_scenario, system_scenario, user_scenario,
};
use integration::iggy_harness;
@@ -99,19 +99,6 @@
quic.keep_alive_interval = "15s"
)
)]
-async fn create_message_payload_scenario(harness: &TestHarness) {
- create_message_payload::run(harness).await;
-}
-
-#[iggy_harness(
- test_client_transport = [Tcp, Http, Quic, WebSocket],
- server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
- quic.max_idle_timeout = "500s",
- quic.keep_alive_interval = "15s"
- )
-)]
async fn stream_size_validation(harness: &TestHarness) {
stream_size_validation_scenario::run(harness).await;
}
diff --git a/core/integration/tests/server/metadata_checkpoint_recovery_vsr.rs b/core/integration/tests/server/metadata_checkpoint_recovery_vsr.rs
deleted file mode 100644
index 4fca67b..0000000
--- a/core/integration/tests/server/metadata_checkpoint_recovery_vsr.rs
+++ /dev/null
@@ -1,138 +0,0 @@
-// 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.
-
-//! Metadata checkpoint-fold recovery across a solo restart, over
-//! `iggy-server-ng`'s production snapshot and WAL path.
-//!
-//! Between checkpoints a replica recovers its metadata by replaying the WAL. Once the
-//! WAL fills, the `SnapshotCoordinator` checkpoints: it persists `snapshot.bin`, pairs
-//! it in the superblock, and DRAINS the snapshotted prefix from the WAL. A restart
-//! after that must fold the snapshot back in as the recovery floor and replay only the
-//! committed suffix on top, rather than rely on a full WAL replay, which no longer
-//! holds the drained ops. This drives a real checkpoint by pushing the metadata WAL
-//! past `CHECKPOINT_MARGIN`, restarts the process, and asserts that a stream from the
-//! drained prefix, recoverable only from the snapshot, and one from the WAL suffix
-//! both survive.
-//!
-//! Solo on purpose: 1-of-1 quorum commits every op the instant it is journaled, so
-//! bulk creation is fast and the WAL is fully committed with no uncommitted suffix to
-//! reconcile, exercising checkpoint and snapshot-fold recovery in isolation without an
-//! election in the mix.
-//!
-//! vsr-only: the metadata snapshot/superblock checkpoint pairing is server-ng's.
-
-use std::time::{Duration, Instant};
-
-use iggy::prelude::*;
-use integration::harness::{TestBinary, TestHarness};
-use integration::iggy_harness;
-use tokio::time::sleep;
-
-/// The prepare WAL has `SLOT_COUNT = 1024` slots and the coordinator checkpoints at
-/// `<= CHECKPOINT_MARGIN` (64) remaining, so after ~960 uncheckpointed ops. 1024
-/// stream creates clears that with room for a WAL suffix above the checkpoint, and
-/// stays under the 4096 stream namespace cap.
-const STREAMS: u32 = 1024;
-
-const RECOVER_TIMEOUT: Duration = Duration::from_secs(60);
-const POLL_INTERVAL: Duration = Duration::from_millis(200);
-
-fn stream_name(index: u32) -> String {
- format!("ckpt-stream-{index}")
-}
-
-#[iggy_harness(cluster_nodes = 1, server(system.sharding.cpu_allocation = "0..1"))]
-async fn given_checkpointed_metadata_when_solo_replica_restarts_should_recover_from_snapshot_and_wal(
- harness: &mut TestHarness,
-) {
- let client = connect(harness).await;
- for index in 0..STREAMS {
- client
- .create_stream(&stream_name(index))
- .await
- .unwrap_or_else(|e| panic!("create stream {index}: {e}"));
- }
- drop(client);
-
- // Crossing CHECKPOINT_MARGIN must have driven the coordinator to persist a snapshot
- // and drain the WAL prefix behind it.
- let snapshot_path = harness
- .node(0)
- .data_path()
- .join("metadata")
- .join("snapshot.bin");
- let snapshot_len = std::fs::metadata(&snapshot_path).map(|m| m.len()).ok();
- assert!(
- snapshot_len.is_some_and(|len| len > 0),
- "{STREAMS} committed metadata ops must cross CHECKPOINT_MARGIN and persist a \
- non-empty snapshot at {}, got {snapshot_len:?}",
- snapshot_path.display()
- );
-
- // Restart the solo node: recovery loads the snapshot, holding the drained prefix,
- // and replays the committed WAL suffix on top.
- harness.node_mut(0).stop().expect("stop the solo node");
- harness.node_mut(0).start().expect("restart the solo node");
-
- // The first stream sits far below the checkpoint op, so it was drained from the WAL
- // and can come back only from the snapshot; the last stream is in the WAL suffix
- // above the checkpoint. Both surviving proves snapshot-fold plus suffix recovery,
- // not a bare WAL replay.
- let client = wait_for_stream(harness, &stream_name(0)).await;
- for name in [stream_name(0), stream_name(STREAMS - 1)] {
- assert!(
- client
- .get_stream(&Identifier::named(&name).unwrap())
- .await
- .expect("get stream after restart")
- .is_some(),
- "stream {name} must survive the checkpointed restart"
- );
- }
-}
-
-/// Connect a root-authenticated TCP client to the solo node.
-async fn connect(harness: &TestHarness) -> IggyClient {
- harness
- .node(0)
- .tcp_client()
- .expect("tcp client builder")
- .with_root_login()
- .connect()
- .await
- .expect("connect to the solo node")
-}
-
-/// Poll the solo node until it is back up and serving `stream`, returning the connected
-/// client. Panics on timeout.
-async fn wait_for_stream(harness: &TestHarness, stream: &str) -> IggyClient {
- let stream_id = Identifier::named(stream).unwrap();
- let deadline = Instant::now() + RECOVER_TIMEOUT;
- loop {
- if let Ok(builder) = harness.node(0).tcp_client()
- && let Ok(client) = builder.with_root_login().connect().await
- && matches!(client.get_stream(&stream_id).await, Ok(Some(_)))
- {
- return client;
- }
- assert!(
- Instant::now() < deadline,
- "solo node did not recover and serve {stream} within {RECOVER_TIMEOUT:?}"
- );
- sleep(POLL_INTERVAL).await;
- }
-}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 5681bf7..8c3f6e0 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -48,10 +48,6 @@
// across a replica restart.
#[cfg(feature = "vsr")]
mod cluster_view_durability_vsr;
-// A metadata checkpoint must drain the WAL and recover from the snapshot fold plus
-// the WAL suffix across a restart.
-#[cfg(feature = "vsr")]
-mod metadata_checkpoint_recovery_vsr;
// 80-case race matrix with hardcoded HTTP variants (test_matrix bypasses
// the harness transport filter).
mod concurrent_addition;
diff --git a/core/integration/tests/server/scenarios/create_message_payload.rs b/core/integration/tests/server/scenarios/create_message_payload.rs
deleted file mode 100644
index 3bb00d1..0000000
--- a/core/integration/tests/server/scenarios/create_message_payload.rs
+++ /dev/null
@@ -1,148 +0,0 @@
-// 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.
-
-use bytes::Bytes;
-use iggy::prelude::*;
-use integration::harness::{TestHarness, assert_clean_system};
-use std::collections::BTreeMap;
-
-const STREAM_NAME: &str = "test-stream";
-const TOPIC_NAME: &str = "test-topic";
-const PARTITIONS_COUNT: u32 = 3;
-const MESSAGES_COUNT: u32 = 1000;
-const PARTITION_ID: u32 = 1;
-
-pub async fn run(harness: &TestHarness) {
- let client = harness
- .root_client()
- .await
- .expect("Failed to get root client");
- init_system(&client).await;
-
- // 1. Send messages with the included headers
- let mut messages = Vec::new();
- for offset in 0..MESSAGES_COUNT {
- let id = (offset + 1) as u128;
- let payload = create_message_payload(offset as u64);
- let headers = create_message_headers();
- let message = IggyMessage::builder()
- .id(id)
- .payload(payload)
- .user_headers(headers)
- .build()
- .expect("Failed to create message with headers");
- messages.push(message);
- }
-
- client
- .send_messages(
- &Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::named(TOPIC_NAME).unwrap(),
- &Partitioning::partition_id(PARTITION_ID),
- &mut messages,
- )
- .await
- .unwrap();
-
- // 2. Poll messages and validate the headers
- let polled_messages = client
- .poll_messages(
- &Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::named(TOPIC_NAME).unwrap(),
- Some(PARTITION_ID),
- &Consumer::default(),
- &PollingStrategy::offset(0),
- MESSAGES_COUNT,
- false,
- )
- .await
- .unwrap();
-
- assert_eq!(polled_messages.messages.len() as u32, MESSAGES_COUNT);
- for i in 0..MESSAGES_COUNT {
- let message = polled_messages.messages.get(i as usize).unwrap();
- assert!(message.user_headers.is_some());
- let headers = message.user_headers_map().unwrap().unwrap();
- assert_eq!(headers.len(), 3);
- assert_eq!(
- headers
- .get(&HeaderKey::try_from("key_1").unwrap())
- .unwrap()
- .as_str()
- .unwrap(),
- "Value 1"
- );
- assert!(
- headers
- .get(&HeaderKey::try_from("key 2").unwrap())
- .unwrap()
- .as_bool()
- .unwrap(),
- );
- assert_eq!(
- headers
- .get(&HeaderKey::try_from("key-3").unwrap())
- .unwrap()
- .as_uint64()
- .unwrap(),
- 123456
- );
- }
- cleanup_system(&client).await;
- assert_clean_system(&client).await;
-}
-
-async fn init_system(client: &IggyClient) {
- // 1. Create the stream
- client.create_stream(STREAM_NAME).await.unwrap();
-
- // 2. Create the topic
- client
- .create_topic(
- &STREAM_NAME.try_into().unwrap(),
- TOPIC_NAME,
- PARTITIONS_COUNT,
- Default::default(),
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .unwrap();
-}
-
-async fn cleanup_system(client: &IggyClient) {
- client
- .delete_stream(&STREAM_NAME.try_into().unwrap())
- .await
- .unwrap();
-}
-
-fn create_message_payload(offset: u64) -> Bytes {
- Bytes::from(format!("message {offset}"))
-}
-
-fn create_message_headers() -> BTreeMap<HeaderKey, HeaderValue> {
- let mut headers = BTreeMap::new();
- headers.insert(
- HeaderKey::try_from("key_1").unwrap(),
- HeaderValue::try_from("Value 1").unwrap(),
- );
- headers.insert(HeaderKey::try_from("key 2").unwrap(), true.into());
- headers.insert(HeaderKey::try_from("key-3").unwrap(), 123456u64.into());
- headers
-}
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 5d701d1..99d20bc 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -28,7 +28,6 @@
pub mod consumer_group_with_multiple_clients_polling_messages_scenario;
pub mod consumer_group_with_single_client_polling_messages_scenario;
pub mod consumer_timestamp_polling_scenario;
-pub mod create_message_payload;
// Cross-protocol PAT visibility (create via HTTP, list via TCP across shards,
// and the reverse). Runs under vsr too: server-ng serves the PAT routes on its
// shard-0 HTTP listener and the create/delete commit through the metadata STM,
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index ce6a86d..6299081 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -87,10 +87,7 @@
disable-mimalloc = []
mimalloc = ["dep:mimalloc"]
iggy-web = ["dep:rust-embed", "dep:mime_guess"]
-# forward iggy/vsr: the SDK gates its VsrSessionControl impls on its own
-# feature while the trait surface follows iggy_common/vsr; without the
-# forward a bare `-p server-ng --features vsr` build breaks in the SDK
-vsr = ["iggy_common/vsr", "iggy/vsr"]
+vsr = ["iggy_common/vsr"]
[dependencies]
ahash = { workspace = true }
@@ -183,7 +180,6 @@
[dev-dependencies]
assert_cmd = { workspace = true }
bytemuck = { workspace = true }
-iggy = { workspace = true }
# Reconciler unit tests assert on `ShardMetrics` snapshots and
# `IggyShard::parked_frame_count`, gated to test/simulator so they cannot grow
# production callers. `shard`'s own `cfg(test)` is false when compiled as our
diff --git a/core/server-ng/tests/sdk_e2e.rs b/core/server-ng/tests/sdk_e2e.rs
index e7f5fcc..2cbd5d6 100644
--- a/core/server-ng/tests/sdk_e2e.rs
+++ b/core/server-ng/tests/sdk_e2e.rs
@@ -15,32 +15,18 @@
// specific language governing permissions and limitations
// under the License.
-//! End-to-end smoke test for the partition reconciliation loop.
+//! Process-level boot smoke test for the `iggy-server-ng` binary.
//!
-//! Spawns the `iggy-server-ng` binary with an isolated tempdir + ephemeral
-//! TCP port, drives an SDK client through the metadata commit path
-//! (`Login` → `CreateStream` → `CreateTopic` with N partitions), then
-//! verifies the per-partition on-disk hierarchy exists. That hierarchy is
-//! created only by
-//! [`server_ng::partition_helpers::create_partition_file_hierarchy`],
-//! inside the reconciler's
-//! [`build_partition_fresh`](server_ng::partition_helpers::build_partition_fresh)
-//! path, so a materialisation observable from outside the process proves
-//! the full commit-notifier → reconciler → disk pipeline.
-//!
-//! Deliberately stops short of `SendMessages` / `PollMessages`: the
-//! partition data-plane SDK round-trip in server-ng is still evolving, so
-//! a smoke test failing for unrelated data-plane reasons would be noisy.
-//! The on-disk hierarchy assertion proves the reconciler ran without
-//! depending on the still-evolving wire surface.
+//! Spawns the production binary against an isolated tempdir with every
+//! transport except TCP disabled and `IGGY_TCP_ADDRESS` bound to port 0,
+//! reads the OS-assigned port back from `runtime/current_config.toml`, and
+//! asserts the listener accepts connections. Scope is the bootstrap path
+//! only: the binary starts, resolves its configuration from the
+//! environment, and reaches a serving state.
use assert_cmd::prelude::CommandCargoExt;
-use iggy::prelude::{
- AutoLogin, Client, CompressionAlgorithm, IggyByteSize, IggyClient, IggyClientBuilder,
- IggyDuration, IggyExpiry, MaxTopicSize, StreamClient, TopicClient, UserClient,
-};
use std::net::{SocketAddr, TcpStream as StdTcpStream};
-use std::path::{Path, PathBuf};
+use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::thread::sleep;
use std::time::{Duration, Instant};
@@ -52,9 +38,9 @@
/// enough for slow CI runners without hanging the suite indefinitely.
const STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
-/// Cadence for polling `current_config.toml` and the partition
-/// directories. Small enough to keep observable latency under a second
-/// without wasting CPU.
+/// Cadence for polling `current_config.toml` while waiting for the
+/// listener to bind. Small enough to keep observable latency under a
+/// second without wasting CPU.
const POLL_INTERVAL: Duration = Duration::from_millis(50);
/// Spawned binary handle. Drops on test exit kill the child + wait so
@@ -80,11 +66,6 @@
.env("IGGY_HTTP_ENABLED", "false")
.env("IGGY_QUIC_ENABLED", "false")
.env("IGGY_WEBSOCKET_ENABLED", "false")
- // Tighten the reconciler tick so the post-CreateTopic
- // materialisation window stays small under CI latency
- // jitter. Hard-coded 200ms is still way above the per-tick
- // re-read cost yet small enough to bound test wall-time.
- .env("IGGY_SYSTEM_SHARDING_RECONCILE_PERIODIC_INTERVAL", "200 ms")
.stdout(Stdio::null())
.stderr(Stdio::piped());
@@ -111,10 +92,6 @@
data_dir,
}
}
-
- fn data_path(&self) -> &Path {
- self.data_dir.path()
- }
}
impl std::fmt::Debug for TestServer {
@@ -181,66 +158,8 @@
Ok(addr)
}
-/// Wait until every per-partition directory has been created by the
-/// reconciler. Returns `Ok(())` once all expected paths exist; surfaces
-/// a structured error on timeout listing which paths were missing.
-fn wait_for_partition_dirs(
- data_dir: &Path,
- stream_id: u32,
- topic_id: u32,
- partition_count: u32,
- timeout: Duration,
-) -> Result<(), String> {
- let expected: Vec<PathBuf> = (0..partition_count)
- .map(|partition_id| partition_dir(data_dir, stream_id, topic_id, partition_id))
- .collect();
- let deadline = Instant::now() + timeout;
- loop {
- let missing: Vec<PathBuf> = expected.iter().filter(|p| !p.exists()).cloned().collect();
- if missing.is_empty() {
- return Ok(());
- }
- if Instant::now() >= deadline {
- return Err(format!(
- "timed out after {timeout:?}; reconciler did not materialise: {missing:?}"
- ));
- }
- sleep(POLL_INTERVAL);
- }
-}
-
-fn partition_dir(data_dir: &Path, stream_id: u32, topic_id: u32, partition_id: u32) -> PathBuf {
- data_dir
- .join("streams")
- .join(stream_id.to_string())
- .join("topics")
- .join(topic_id.to_string())
- .join("partitions")
- .join(partition_id.to_string())
-}
-
-async fn connected_client(server_addr: SocketAddr) -> IggyClient {
- let client = IggyClientBuilder::new()
- .with_tcp()
- .with_server_address(server_addr.to_string())
- .with_auto_sign_in(AutoLogin::Disabled)
- .with_reconnection_max_retries(Some(3))
- .with_reconnection_interval(IggyDuration::from(200_000_u64))
- .build()
- .expect("TCP client build");
- client.connect().await.expect("SDK TCP connect");
- client
- .login_user("iggy", "iggy")
- .await
- .expect("SDK login as root");
- client
-}
-
/// Spawn `iggy-server-ng` and verify the production binary bootstraps
-/// cleanly. Proves the test infrastructure (binary path, env-var
-/// overrides, runtime config discovery) is wired correctly so the
-/// `#[ignore]`d full end-to-end test below can be flipped on the
-/// moment the wire bridge lands.
+/// cleanly.
///
/// Specifically asserts:
/// * The binary path is reachable via `cargo_bin`.
@@ -263,73 +182,3 @@
let _ = StdTcpStream::connect_timeout(&server.tcp_addr, Duration::from_secs(1))
.expect("server-ng TCP listener must accept connections post-bootstrap");
}
-
-/// End-to-end smoke test against a live `iggy-server-ng` binary.
-///
-/// Currently gated `#[ignore]` because server-ng's client-facing wire
-/// surface is not yet SDK-compatible: the SDK's TCP framing (length-
-/// prefixed binary, see `core/sdk/src/tcp/tcp_client.rs`) does not match
-/// the server-ng client listener which speaks the consensus
-/// [`RequestHeader`] / [`PrepareHeader`] frame layout. The TCP socket
-/// accepts, but `login_user` either hangs on the framing read or
-/// retries until the SDK surfaces
-/// [`iggy_common::IggyError::CannotEstablishConnection`].
-///
-/// The structural pieces this test exercises are still useful as
-/// infrastructure once the wire bridge lands (tracked as a follow-up):
-/// * tempdir + ephemeral-port spawn of the production binary
-/// * `runtime/current_config.toml` port discovery
-/// * SDK-driven `CreateStream` + `CreateTopic` round-trip
-/// * on-disk hierarchy assertion proving the reconciler
-/// materialised every partition end-to-end
-///
-/// Flip the `#[ignore]` to active once `iggy-server-ng` accepts SDK
-/// frames; no other changes should be needed.
-#[ignore = "server-ng client wire surface is not yet SDK-compatible"]
-#[tokio::test(flavor = "current_thread")]
-async fn reconciler_materialises_partitions_on_create_topic_e2e() {
- let server = TestServer::start();
- let client = connected_client(server.tcp_addr).await;
-
- // CreateStream commits a `CreateStream` op; the metadata STM
- // assigns slab key 0 (first stream on a fresh server). The notifier
- // does NOT broadcast a `MetadataCommitTick` for plain stream commits
- // (the reconciler only cares about partition-shaped ops) but the
- // call validates the wire is reachable before we move on.
- client
- .create_stream("e2e-stream")
- .await
- .expect("create stream");
-
- // CreateTopic commits as `CreateTopicWithAssignments` after the
- // primary's allocator stamps consensus_group_id values; the
- // notifier broadcasts `MetadataCommitTick`, the reconciler wakes,
- // and `build_partition_fresh` calls
- // `create_partition_file_hierarchy` for each assigned partition.
- let partitions: u32 = 3;
- client
- .create_topic(
- &"e2e-stream".try_into().expect("stream identifier"),
- "e2e-topic",
- partitions,
- CompressionAlgorithm::default(),
- None,
- IggyExpiry::ServerDefault,
- MaxTopicSize::Custom(IggyByteSize::from(1024_u64 * 1024 * 1024)),
- )
- .await
- .expect("create topic");
-
- // Reconciler is asynchronous: wait for the on-disk hierarchy to
- // appear. The 200ms periodic tick configured in `TestServer::start`
- // bounds the worst-case wait to ~two ticks even if the post-commit
- // wake-up frame got coalesced.
- wait_for_partition_dirs(
- server.data_path(),
- /* stream_id */ 0,
- /* topic_id */ 0,
- partitions,
- Duration::from_secs(5),
- )
- .expect("reconciler materialised all partition directories");
-}
