core/integration/tests/server/specific.rs

// 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 crate::server::scenarios::{message_size_scenario, single_message_per_batch_scenario};
use crate::server::scenarios::{reconnect_after_restart_scenario, restart_offset_skip_scenario};
use crate::server::scenarios::{
    segment_rotation_race_scenario, tcp_tls_scenario, websocket_tls_scenario,
};
use integration::iggy_harness;

#[iggy_harness(
    test_client_transport = TcpTlsGenerated,
    server(tls = generated)
)]
async fn tcp_tls_scenario_should_be_valid(harness: &TestHarness) {
    let client = harness.root_client().await.unwrap();
    tcp_tls_scenario::run(&client).await;
}

#[iggy_harness(
    test_client_transport = TcpTlsSelfSigned,
    server(tls = self_signed)
)]
async fn tcp_tls_self_signed_scenario_should_be_valid(harness: &TestHarness) {
    let client = harness.root_client().await.unwrap();
    tcp_tls_scenario::run(&client).await;
}

#[iggy_harness(
    test_client_transport = WebSocketTlsGenerated,
    server(websocket_tls = generated)
)]
async fn websocket_tls_scenario_should_be_valid(harness: &TestHarness) {
    let client = harness.root_client().await.unwrap();
    websocket_tls_scenario::run(&client).await;
}

#[iggy_harness]
async fn message_size_scenario(harness: &TestHarness) {
    message_size_scenario::run(harness).await;
}

#[iggy_harness]
async fn should_handle_single_message_per_batch_with_delayed_persistence(harness: &TestHarness) {
    single_message_per_batch_scenario::run(harness, 5).await;
}

#[iggy_harness(
    test_client_transport = [Tcp, WebSocket, Quic],
    server(
        quic.max_idle_timeout = "500s",
        quic.keep_alive_interval = "15s"
    )
)]
async fn producer_reconnect_after_server_restart(harness: &mut TestHarness) {
    reconnect_after_restart_scenario::run_producer(harness).await;
}

// QUIC is excluded on an SDK gap: after the restart the QUIC client redirects
// to the new leader, reconnects, and signs in, but the long-lived consumer's
// polls then return nothing for the whole window -- the post-reconnect request
// path wedges (QUIC also lacks the TCP client's mid-connection failover). TCP
// and WebSocket run.
#[iggy_harness(
    test_client_transport = [Tcp, WebSocket],
    server(
        quic.max_idle_timeout = "500s",
        quic.keep_alive_interval = "15s"
    )
)]
async fn consumer_reconnect_after_server_restart(harness: &mut TestHarness) {
    reconnect_after_restart_scenario::run_consumer(harness).await;
}

#[iggy_harness]
async fn single_message_restart_offset_zero(harness: &mut TestHarness) {
    reconnect_after_restart_scenario::run_single_message_offset_zero_restart(harness).await;
}

// Exercises the rejoin probe's election fallback across all replicas, which a
// plain single-node restart does not reach.
#[iggy_harness]
async fn full_cluster_restart_recovers_and_serves(harness: &mut TestHarness) {
    reconnect_after_restart_scenario::run_full_cluster_restart(harness).await;
}

// Exercises `RangeEvicted` + the commit floor: the rejoin window exceeds the
// peers' evicted ring, so journal repair alone cannot cover it.
#[iggy_harness]
async fn rejoin_window_exceeding_evicted_ring(harness: &mut TestHarness) {
    reconnect_after_restart_scenario::run_ring_overflow_rejoin(harness).await;
}

#[iggy_harness]
async fn consumer_offset_ahead_after_crash(harness: &mut TestHarness) {
    reconnect_after_restart_scenario::run_consumer_offset_ahead_after_crash(harness).await;
}

/// Regression test: consumer offset skip after server restart during concurrent
/// produce+consume. Reproduces the exact scenario from issue #2924/#2715:
/// send messages, restart server, produce+consume concurrently, verify no offset
/// gaps.
///
/// Config: high messages_required_to_save so post-restart messages accumulate in
/// the journal (exposing the base_offset=0 bug).
#[iggy_harness]
async fn restart_offset_skip(harness: &mut TestHarness) {
    restart_offset_skip_scenario::run(harness).await;
}

/// This test configures the server to trigger frequent segment rotations and runs
/// multiple concurrent producers across all protocols (TCP, HTTP, QUIC, WebSocket)
/// to maximize the chance of hitting the race condition between persist_messages_to_disk
/// and handle_full_segment.
///
/// Server configuration:
/// - Smallest segment size a topic may declare (1 MiB), plus a payload sized
///   to keep rotations frequent at that floor (~240 rolls per run)
/// - Small messages_required_to_save (32) to trigger more frequent saves
///
/// Test configuration:
/// - 8 producers total (2 per protocol: TCP, HTTP, QUIC, WebSocket)
/// - All producers write to the same partition for maximum lock contention
// Concurrency race test: runs over the three VSR transports (TCP/QUIC/
// WebSocket -- HTTP/REST carries no VSR framing).
#[iggy_harness]
async fn segment_rotation_scenario(harness: &TestHarness) {
    segment_rotation_race_scenario::run(harness).await;
}