This is an automated email from the ASF dual-hosted git repository.
spetz 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 01a64b2e0 test(integration): fix vsr flakes from racing recovery
routes (#3833)
01a64b2e0 is described below
commit 01a64b2e00eb3037d45a85e4cdefe09eed6e2e07
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Sat Aug 8 08:38:14 2026 +0200
test(integration): fix vsr flakes from racing recovery routes (#3833)
---
.../tests/cluster/metadata_state_transfer.rs | 28 ++-
.../tests/cluster/partition_state_transfer.rs | 67 ++++--
.../integration/tests/connectors/http/http_sink.rs | 65 ++---
core/integration/tests/data_integrity/mod.rs | 11 +-
.../verify_consumer_group_partition_assignment.rs | 18 +-
.../server/scenarios/message_cleanup_scenario.rs | 129 ++++++----
core/integration/tests/server/scenarios/mod.rs | 2 -
.../scenarios/stress_produce_consume_scenario.rs | 264 ---------------------
core/partitions/src/iggy_partition.rs | 13 +
9 files changed, 226 insertions(+), 371 deletions(-)
diff --git a/core/integration/tests/cluster/metadata_state_transfer.rs
b/core/integration/tests/cluster/metadata_state_transfer.rs
index b66bfacf7..275946412 100644
--- a/core/integration/tests/cluster/metadata_state_transfer.rs
+++ b/core/integration/tests/cluster/metadata_state_transfer.rs
@@ -301,21 +301,33 @@ async fn
given_election_past_a_node_when_it_rejoins_stale_should_probe_then_stat
// while the cluster sits at a higher view -- the stale-primary case.
harness.restart_node_from_clean_slate(0).unwrap();
- // It must first PROBE (its heartbeat-send timer converting, since it has
no
- // heartbeat-receive timer as a "primary"), then complete the transfer. The
- // probe marker distinguishes this path from the same-view backstop.
+ // It must leave view 0 before it can transfer, and there are two correct
+ // routes off it. Either its heartbeat-SEND timer converts into a probe (it
+ // has no heartbeat-RECEIVE timer while it believes itself primary), or an
+ // unsolicited `StartView` from the live primary reaches it first and it
+ // adopts that. Which one wins is scheduler luck: the probe timer races the
+ // survivors' next StartView broadcast, and on an unloaded box the
StartView
+ // lands ~450ms into boot and takes it. Pinning the probe marker alone made
+ // this spec fail whenever the node caught up the faster way. What must
hold
+ // either way is that it left the stale view by a legitimate route and then
+ // converged rather than wedging.
let deadline = Instant::now() + TRANSFER_BUDGET;
- let mut probed = false;
+ let mut left_stale_view = false;
loop {
- probed = probed || harness.node(0).stdout_contains("probing to catch
up");
+ left_stale_view = left_stale_view
+ || harness.node(0).stdout_contains("probing to catch up")
+ || harness
+ .node(0)
+ .stdout_contains("adopting view from StartView");
if harness
.node(0)
.stdout_contains("metadata state transfer installed")
{
assert!(
- probed,
- "the rejoined node transferred without first probing; the \
- stale-view path must reach the transfer through a view probe"
+ left_stale_view,
+ "the rejoined node transferred while still believing view 0;
it \
+ must first leave the stale view, by its own probe or by
adopting \
+ a StartView"
);
break;
}
diff --git a/core/integration/tests/cluster/partition_state_transfer.rs
b/core/integration/tests/cluster/partition_state_transfer.rs
index 0d3647917..13413d886 100644
--- a/core/integration/tests/cluster/partition_state_transfer.rs
+++ b/core/integration/tests/cluster/partition_state_transfer.rs
@@ -68,6 +68,15 @@ const INSTALL_MARKER: &str = "partition state transfer
installed";
const FULLY_SERVED_MARKER: &str = "partition state transfer fully served";
const ABANDON_MARKER: &str =
"partition state transfer stalled past its retry budget; abandoning with a
backed-off re-arm";
+/// The second route off a dead transfer peer. A replica parked in
+/// `AwaitingTarget` re-arms the moment it adopts a StartView, and with the
+/// stock config that beats the stall budget every time: the budget needs six
+/// rounds of `repair_retry_interval` (~6s) while `heartbeat_timeout` elects
+/// the new primary in 5s. Both routes prove the same thing - the replica did
+/// not retry into the corpse - so a spec that pins one is asserting on
+/// scheduler luck.
+const REARM_ON_VIEW_MARKER: &str =
+ "adopted a live view while awaiting transfer; requesting partition state
transfer";
/// Transfer end-to-end: adoption, repair round-trip, conversion, chunk pull,
/// install, tail repair. CI runners are slow; bound without hanging the suite.
@@ -243,7 +252,7 @@ async fn
given_evicted_ring_when_node_restarts_with_data_should_state_transfer_p
// pull also runs the staged-segment reuse scan, but how much it can adopt
// depends on how many artifacts completed before the kill, so nothing here
// asserts on reuse.
-async fn
given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partition(
+async fn
given_transfer_peer_dies_when_stalled_should_leave_dead_peer_and_recover_partition(
harness: &mut TestHarness,
) {
let client = harness
@@ -254,26 +263,26 @@ async fn
given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partit
// Bulky payloads so the pull spans many 256 KiB chunks: the kill below
// must land while the transfer is provably in flight, and a small
// partition finishes inside the marker-poll latency, leaving the
- // abandon path untested.
+ // dead-peer path untested.
produce_bulky(&client, BULKY_MESSAGES_COUNT, BULKY_PAYLOAD_LEN).await;
sleep(Duration::from_secs(1)).await;
let _seed_client = client;
- // Wipe node 2, wait until its rejoin CONVERTED to a transfer and the
- // serving peer (the view-0 primary, node 0) proved it started serving
- // the pull, then kill that peer mid-pull. Node 2 must not retry into
- // the corpse forever: the stall budget abandons with a backed-off
- // re-arm against the next peer, the survivors elect past node 0, and
- // the transfer re-runs against the new primary (node 1).
+ // Wipe node 2, wait until its rejoin CONVERTED to a transfer, then kill
+ // the serving peer (the view-0 primary, node 0) mid-pull. Node 2 must not
+ // retry into the corpse forever: it re-arms against the next peer -- via
+ // the stall budget, or via the StartView it adopts once the survivors
+ // elect past node 0 -- and the transfer re-runs against the new primary
+ // (node 1).
harness
.restart_node_from_clean_slate(2)
.expect("clean-slate restart of node 2");
// Kill node 0 the moment node 2 CONVERTED: the transfer is then armed
// at node 0 but the 64 MiB pull cannot possibly finish inside the kill
- // latency, so node 2 deterministically ends up stalling against a dead
- // peer -- whether the descriptor made it out or not, both funnels land
- // in the stall budget. (Gating on the serving marker instead raced the
- // pull itself: a release-build pull finishes in a few hundred ms.)
+ // latency, so node 2 deterministically ends up parked against a dead
+ // peer, whether the descriptor made it out or not. (Gating on the serving
+ // marker instead raced the pull itself: a release-build pull finishes in
+ // a few hundred ms.)
let deadline = Instant::now() + TRANSFER_BUDGET;
while !harness.node(2).stdout_contains(CONVERSION_MARKER) {
assert!(
@@ -289,13 +298,22 @@ async fn
given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partit
let installs_before_kill =
harness.node(2).stdout_occurrences(INSTALL_MARKER);
let served_before_kill =
harness.node(1).stdout_occurrences(FULLY_SERVED_MARKER);
let abandons_before_kill =
harness.node(2).stdout_occurrences(ABANDON_MARKER);
+ let rearms_before_kill =
harness.node(2).stdout_occurrences(REARM_ON_VIEW_MARKER);
harness
.stop_node(0)
.expect("stop the serving peer (node 0)");
- // The abandon is now deterministic: the pull was in flight against a
- // peer that is gone, so the stall budget must exhaust.
- await_new_marker(harness, 2, ABANDON_MARKER, abandons_before_kill).await;
+ // The pull was in flight against a peer that is gone, so node 2 must leave
+ // the dead session by one of the two routes off it, whichever fires first.
+ await_new_marker_any(
+ harness,
+ 2,
+ &[
+ (ABANDON_MARKER, abandons_before_kill),
+ (REARM_ON_VIEW_MARKER, rearms_before_kill),
+ ],
+ )
+ .await;
// Recovery: the scheduled re-arm targets the surviving primary. No
// follow-up commit is asserted -- the cluster is quorum-marginal with
@@ -497,6 +515,25 @@ async fn await_new_marker(harness: &TestHarness, node:
usize, marker: &str, base
}
}
+/// [`await_new_marker`] over alternative markers on one node, each carried
+/// with its own pre-fault baseline: satisfied by the first to advance.
+async fn await_new_marker_any(harness: &TestHarness, node: usize, markers:
&[(&str, usize)]) {
+ let deadline = Instant::now() + TRANSFER_BUDGET;
+ loop {
+ if markers
+ .iter()
+ .any(|(marker, baseline)|
harness.node(node).stdout_occurrences(marker) > *baseline)
+ {
+ return;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "node {node} logged none of {markers:?} again after the fault
within {TRANSFER_BUDGET:?}"
+ );
+ sleep(MARKER_POLL).await;
+ }
+}
+
/// [`await_marker`] over a set of nodes: satisfied by the first one to log it.
async fn await_marker_any(harness: &TestHarness, nodes: &[usize], marker:
&str) {
let deadline = Instant::now() + TRANSFER_BUDGET;
diff --git a/core/integration/tests/connectors/http/http_sink.rs
b/core/integration/tests/connectors/http/http_sink.rs
index 7e20c25f8..0466e71f4 100644
--- a/core/integration/tests/connectors/http/http_sink.rs
+++ b/core/integration/tests/connectors/http/http_sink.rs
@@ -301,41 +301,36 @@ async fn
individual_json_messages_delivered_as_separate_posts(
// Test 2: NDJSON Batch Mode
// ============================================================================
-/// Validates `batch_mode=ndjson`: all messages in one request as
newline-delimited JSON.
-/// Checks single request, line count = message count, per-line envelope,
`application/x-ndjson`.
-#[iggy_harness(
- server(connectors_runtime(config_path =
"tests/connectors/http/sink.toml")),
- seed = seeds::connector_stream
-)]
-async fn ndjson_messages_delivered_as_single_request(
- harness: &TestHarness,
- fixture: HttpSinkNdjsonFixture,
-) {
- let client = harness.root_client().await.unwrap();
- let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
- let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+/// Creates the connector stream/topic and pre-publishes the NDJSON test
messages.
+///
+/// Same reason as [`connector_stream_with_json_array_messages`]: publishing
from the
+/// test body races the sink's poll loop against the server's commit frontier,
so a
+/// poll can observe a partial batch and flush it as its own request. Messages
at rest
+/// before the first poll always arrive as one batch.
+async fn connector_stream_with_ndjson_messages(
+ client: &IggyClient,
+) -> Result<(), seeds::SeedError> {
+ seeds::connector_stream(client).await?;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into()?;
+ let topic_id: Identifier = seeds::names::TOPIC.try_into()?;
- // Step 1: Build 3 JSON event messages
let json_payloads: Vec<serde_json::Value> = vec![
serde_json::json!({"event": "login", "user": 1}),
serde_json::json!({"event": "click", "user": 2}),
serde_json::json!({"event": "logout", "user": 3}),
];
- let mut messages: Vec<IggyMessage> = json_payloads
- .iter()
- .enumerate()
- .map(|(i, payload)| {
- let bytes = serde_json::to_vec(payload).expect("Failed to
serialize");
+ let mut messages: Vec<IggyMessage> =
Vec::with_capacity(json_payloads.len());
+ for (i, payload) in json_payloads.iter().enumerate() {
+ messages.push(
IggyMessage::builder()
.id((i + 1) as u128)
- .payload(Bytes::from(bytes))
- .build()
- .expect("Failed to build message")
- })
- .collect();
+ .payload(Bytes::from(serde_json::to_vec(payload)?))
+ .build()?,
+ );
+ }
- // Step 2: Publish messages to Iggy
client
.send_messages(
&stream_id,
@@ -343,10 +338,20 @@ async fn ndjson_messages_delivered_as_single_request(
&Partitioning::partition_id(0),
&mut messages,
)
- .await
- .expect("Failed to send messages");
+ .await?;
+
+ Ok(())
+}
- // Step 3: Wait for single NDJSON request (all messages batched into one)
+/// Validates `batch_mode=ndjson`: all messages in one request as
newline-delimited JSON.
+/// Checks single request, line count = message count, per-line envelope,
`application/x-ndjson`.
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/http/sink.toml")),
+ seed = connector_stream_with_ndjson_messages
+)]
+async fn ndjson_messages_delivered_as_single_request(fixture:
HttpSinkNdjsonFixture) {
+ // Step 1: Wait for the single NDJSON request. The seed pre-published all
+ // messages, so the sink delivers them in one batch (see the seed docs
above).
let requests = fixture
.container()
.wait_for_requests(1)
@@ -357,7 +362,7 @@ async fn ndjson_messages_delivered_as_single_request(
assert_eq!(req.method, "POST", "Expected POST method");
assert_eq!(req.url, "/ingest", "Expected /ingest URL");
- // Step 4: Parse NDJSON body — each line is a separate JSON envelope
+ // Step 2: Parse NDJSON body — each line is a separate JSON envelope
let lines: Vec<&str> = req.body.trim().lines().collect();
assert_eq!(
lines.len(),
@@ -379,7 +384,7 @@ async fn ndjson_messages_delivered_as_single_request(
);
}
- // Step 5: Verify NDJSON content type
+ // Step 3: Verify NDJSON content type
let ct = req
.header("Content-Type")
.expect("Content-Type header must be present");
diff --git a/core/integration/tests/data_integrity/mod.rs
b/core/integration/tests/data_integrity/mod.rs
index 620f8e872..5a3815d2d 100644
--- a/core/integration/tests/data_integrity/mod.rs
+++ b/core/integration/tests/data_integrity/mod.rs
@@ -15,10 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-// Partially vsr-gated inside the module: the bench-fill test requires
-// PARTITION-plane state transfer (sub-floor stats/offset seeding), which is
-// not implemented yet; the metadata-only deletion/restart test runs under
-// vsr (metadata journal repair covers its rejoin window).
+// Partially vsr-gated inside the module: the remaining gates cover
+// `flush_unsaved_buffer`, which server-ng answers `FeatureUnavailable` and
+// which the eager-flush server envs replace under vsr. The bench-fill test
+// itself runs under vsr since PARTITION-plane state transfer landed, but the
+// harness spawns `iggy-bench` off disk with no cargo build-graph edge, so the
+// binary must have been built `--features vsr` or its login hangs on the
+// framing mismatch.
mod verify_after_server_restart;
mod verify_user_login_after_restart;
diff --git
a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
index f5f378837..b92c5bcf4 100644
---
a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
+++
b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
@@ -28,6 +28,11 @@ const STREAM_NAME: &str = "cg-partition-test-stream";
const TOPIC_NAME: &str = "cg-partition-test-topic";
const CONSUMER_GROUP_NAME: &str = "cg-partition-test-group";
const PARTITIONS_COUNT: u32 = 3;
+/// Slices the slab-reuse wait so the surviving consumer can prove liveness
+/// inside the server's staleness window. Product of the two is the 3s the
+/// spec waits for the freed slab to become reusable.
+const CONSUMER1_KEEPALIVE_PINGS: u32 = 3;
+const CONSUMER1_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1);
async fn create_stale_tcp_client(server_addr: &str) -> IggyClient {
let config = TcpClientConfig {
@@ -3144,7 +3149,18 @@ async fn
should_not_assign_partition_to_wrong_member_after_slab_reuse(harness: &
// 3. Consumer2 (revocation target) disconnects — its slab is freed
drop(client2);
- sleep(Duration::from_secs(3)).await;
+ // Consumer1 must stay alive across the wait. This spec asks the server for
+ // `heartbeat.interval = 2s`, so its verifier evicts any consumer-group
+ // member idle past 1.2 intervals, and harness clients never ping on their
+ // own (the SDK pinger is spawned by `IggyClient::connect`, which the
+ // builder does not call). Silence here evicted consumer1 mid-wait and its
+ // offset store below came back `StaleClient`. Any request refreshes
+ // liveness; a ping is the cheapest. Consumer2 stays silent by construction
+ // - it is already dropped, and the socket close frees its slab.
+ for _ in 0..CONSUMER1_KEEPALIVE_PINGS {
+ sleep(CONSUMER1_KEEPALIVE_INTERVAL).await;
+ client1.ping().await.unwrap();
+ }
// 4. Consumer3 joins — may reuse consumer2's old slab
let client3 = harness.new_client().await.unwrap();
diff --git
a/core/integration/tests/server/scenarios/message_cleanup_scenario.rs
b/core/integration/tests/server/scenarios/message_cleanup_scenario.rs
index af2519729..51e67d6a4 100644
--- a/core/integration/tests/server/scenarios/message_cleanup_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_cleanup_scenario.rs
@@ -17,9 +17,9 @@
//! Tests for message retention policies (time-based and size-based).
//!
-//! Configuration: 100KB segment size, 100ms cleaner interval, instant flush.
+//! Configuration: 10KiB segment size, 100ms cleaner interval, instant flush.
//! Message size: 64B header + 936B payload = 1KB per message.
-//! Therefore: 100 messages = 1 segment, 101+ messages = 2+ segments.
+//! Therefore a segment holds 9 messages and every ~10 messages rotates one.
use bytes::Bytes;
use iggy::prelude::*;
@@ -48,7 +48,13 @@ pub async fn run_expiry_after_rotation(client: &IggyClient,
data_path: &Path) {
let stream = client.create_stream(STREAM_NAME).await.unwrap();
let stream_id = stream.id;
- let expiry = Duration::from_secs(2);
+ // The whole send burst has to land inside this window: expiry runs off
each
+ // message's own timestamp, so a produce run that outlives it gets its
+ // oldest segments reclaimed before the pre-expiry poll ever runs. A 3-node
+ // vsr cluster in a debug build pays a consensus round-trip plus an fsync
+ // per request, which is what pushed the old one-message-per-request loop
+ // past the old 2s window.
+ let expiry = Duration::from_secs(4);
let topic = client
.create_topic(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -70,18 +76,24 @@ pub async fn run_expiry_after_rotation(client: &IggyClient,
data_path: &Path) {
.display()
.to_string();
- // Send 110 messages (1KB each) to create 2 segments (100KB segment size)
+ // Send 110 messages (1KB each) in batches, spanning several 10KiB
segments.
+ // Batched rather than one request per message: the burst must fit inside
+ // `expiry` with room to spare, and each request costs a round-trip.
let payload = make_payload('A');
- let total_messages = 110;
-
- for i in 0..total_messages {
- let message = IggyMessage::builder()
- .id(i as u128)
- .payload(payload.clone())
- .build()
- .unwrap();
-
- let mut messages = vec![message];
+ let total_messages: usize = 110;
+ let batch_size = 10;
+
+ for chunk_start in (0..total_messages).step_by(batch_size) {
+ let mut messages: Vec<IggyMessage> = (chunk_start
+ ..total_messages.min(chunk_start + batch_size))
+ .map(|i| {
+ IggyMessage::builder()
+ .id(i as u128)
+ .payload(payload.clone())
+ .build()
+ .unwrap()
+ })
+ .collect();
client
.send_messages(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -117,7 +129,7 @@ pub async fn run_expiry_after_rotation(client: &IggyClient,
data_path: &Path) {
assert_eq!(
polled_before.messages.len(),
- total_messages as usize,
+ total_messages,
"Should poll all messages before expiry"
);
@@ -345,16 +357,24 @@ pub async fn run_combined_retention(client: &IggyClient,
data_path: &Path) {
.display()
.to_string();
- // Send 110 messages to create 2 segments (under size threshold, but will
expire)
+ // Send 110 messages to create 2 segments (under size threshold, but will
+ // expire). Batched so the burst finishes well inside `expiry`: a
per-message
+ // request pays a consensus round-trip plus an fsync, and a loop that
+ // outlives the window has its head reclaimed before the count below.
let payload = make_payload('C');
- for i in 0..110 {
- let message = IggyMessage::builder()
- .id(i as u128)
- .payload(payload.clone())
- .build()
- .unwrap();
-
- let mut messages = vec![message];
+ let total_messages: usize = 110;
+ let batch_size = 10;
+ for chunk_start in (0..total_messages).step_by(batch_size) {
+ let mut messages: Vec<IggyMessage> = (chunk_start
+ ..total_messages.min(chunk_start + batch_size))
+ .map(|i| {
+ IggyMessage::builder()
+ .id(i as u128)
+ .payload(payload.clone())
+ .build()
+ .unwrap()
+ })
+ .collect();
client
.send_messages(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -412,19 +432,25 @@ pub async fn run_expiry_with_multiple_partitions(client:
&IggyClient, data_path:
let topic_id = topic.id;
let payload = make_payload('D');
- let messages_per_partition = 110;
+ let messages_per_partition: usize = 110;
+ let batch_size = 10;
- // Send messages to all partitions
+ // Send messages to all partitions. Batched: a per-message request costs a
+ // consensus round-trip plus an fsync, and `PARTITIONS_COUNT` × 110 of
those
+ // outlive `expiry`, so the cleaner would reclaim the first partition's
+ // sealed segments before the last one had even been written.
for partition_id in 0..PARTITIONS_COUNT {
- for i in 0..messages_per_partition {
- let msg_id = partition_id as u128 * 1000 + i as u128;
- let message = IggyMessage::builder()
- .id(msg_id)
- .payload(payload.clone())
- .build()
- .unwrap();
-
- let mut messages = vec![message];
+ for chunk_start in (0..messages_per_partition).step_by(batch_size) {
+ let mut messages: Vec<IggyMessage> = (chunk_start
+ ..messages_per_partition.min(chunk_start + batch_size))
+ .map(|i| {
+ IggyMessage::builder()
+ .id(partition_id as u128 * 1000 + i as u128)
+ .payload(payload.clone())
+ .build()
+ .unwrap()
+ })
+ .collect();
client
.send_messages(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -594,10 +620,11 @@ pub async fn run_expiry_respects_consumer_offset(client:
&IggyClient, data_path:
let stream = client.create_stream(TEST_STREAM).await.unwrap();
let stream_id = stream.id;
- // Expiry must outlast the send + first-poll phase: 300 serial sends with
- // per-message fsync (and VSR quorum in cluster mode) take ~3s under load.
- // If segments expire before the consumer commits its first offset, there
is
- // no barrier yet and the cleaner legally deletes them, breaking the
premise.
+ // Expiry must outlast the send + first-poll phase. If segments expire
+ // before the consumer commits its first offset, there is no barrier yet
+ // and the cleaner legally deletes them, breaking the premise: the poll
+ // below then starts at the earliest surviving offset instead of 0.
+ // The sends are batched for the same reason (see below).
let expiry = Duration::from_secs(4);
let topic = client
.create_topic(
@@ -620,21 +647,29 @@ pub async fn run_expiry_respects_consumer_offset(client:
&IggyClient, data_path:
.display()
.to_string();
- // Send 300 messages (1KB each) -> 3 sealed segments + active
+ // Send 300 messages (1KB each) -> 3 sealed segments + active. Batched:
+ // one request per message costs a consensus round-trip plus an fsync each,
+ // which on a 3-node debug cluster runs the burst well past `expiry`.
let payload = make_payload('B');
let total_messages = 300u32;
- for i in 0..total_messages {
- let message = IggyMessage::builder()
- .id(i as u128)
- .payload(payload.clone())
- .build()
- .unwrap();
+ let batch_size = 10u32;
+ for chunk_start in (0..total_messages).step_by(batch_size as usize) {
+ let mut messages: Vec<IggyMessage> = (chunk_start
+ ..total_messages.min(chunk_start + batch_size))
+ .map(|i| {
+ IggyMessage::builder()
+ .id(i as u128)
+ .payload(payload.clone())
+ .build()
+ .unwrap()
+ })
+ .collect();
client
.send_messages(
&Identifier::named(TEST_STREAM).unwrap(),
&Identifier::named(TEST_TOPIC).unwrap(),
&Partitioning::partition_id(PARTITION_ID),
- &mut [message],
+ &mut messages,
)
.await
.unwrap();
diff --git a/core/integration/tests/server/scenarios/mod.rs
b/core/integration/tests/server/scenarios/mod.rs
index 99d20bcf8..5287c3053 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -50,8 +50,6 @@ pub mod single_message_per_batch_scenario;
pub mod snapshot_scenario;
pub mod stale_client_consumer_group_scenario;
pub mod stream_size_validation_scenario;
-#[cfg(feature = "vsr")]
-pub mod stress_produce_consume_scenario;
pub mod system_scenario;
pub mod tcp_tls_scenario;
pub mod timestamp_scenario;
diff --git
a/core/integration/tests/server/scenarios/stress_produce_consume_scenario.rs
b/core/integration/tests/server/scenarios/stress_produce_consume_scenario.rs
deleted file mode 100644
index 507df4300..000000000
--- a/core/integration/tests/server/scenarios/stress_produce_consume_scenario.rs
+++ /dev/null
@@ -1,264 +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.
-
-//! Data-plane produce+consume stress across server-ng topologies: `PRODUCERS`
-//! producers and `CONSUMERS` consumers all hammer a SINGLE partition,
asserting
-//! no message loss and a strictly contiguous offset log.
-//!
-//! Targets the partition-ref-across-await UB fix: the consume poll path and
the
-//! produce/commit pump run as sibling tasks over the same partition, so
-//! concentrating every producer and consumer on one partition maximizes the
-//! `&`/`&mut` aliasing window on that partition's pump that the fix closes.
All
-//! producers run concurrently with all consumers for `HAMMER_DURATION`; the
-//! consumers then drain. A single partition lives on a single shard, so the
-//! multi-shard variants still spin up N shards but concentrate the load on the
-//! one owning shard.
-//!
-//! Oracle: producers interleave on the partition's shared offset sequence, so
-//! per-producer contiguity does not hold. Instead every consumer reads the
-//! partition in full and must observe a contiguous `0..total` (no gap = no
loss,
-//! no dup) with a count equal to the sum of all producers' sends.
-//!
-//! Strictly data plane: polls by explicit offset with `auto_commit = false`
and
-//! performs no mid-run topic/partition mutation, so it never drives the
metadata
-//! consensus plane concurrently. That avoids a separate, still-open `on_ack`
-//! journal-durability race that panics the primary under concurrent metadata
ops
-//! (see the gated `concurrent_produce_consume_scenario` in
`scenarios/mod.rs`).
-
-use bytes::Bytes;
-use iggy::prelude::*;
-use integration::harness::TestHarness;
-use integration::iggy_harness;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::time::{Duration, Instant};
-
-const STREAM_NAME: &str = "stress-pc-stream";
-const TOPIC_NAME: &str = "stress-pc-topic";
-// All traffic targets this one partition to concentrate produce+consume
-// contention on a single partition pump.
-const PARTITION_ID: u32 = 0;
-const PRODUCERS: u32 = 4;
-const CONSUMERS: u32 = 4;
-const PRODUCER_BATCH: u32 = 16;
-const CONSUMER_BATCH: u32 = 64;
-const HAMMER_DURATION: Duration = Duration::from_secs(20);
-// Safety net so a wedged consumer fails loudly instead of hanging the suite.
-const MAX_TEST_DURATION: Duration = Duration::from_secs(120);
-// Whole-test wall-clock guard. A server that dies at boot leaves the harness
-// client retrying connect with no cap, and a parked poll never re-checks
-// MAX_TEST_DURATION, so without this the suite hangs indefinitely instead of
-// failing. Set above MAX_TEST_DURATION so a slow-but-progressing consumer
-// trips its own informative deadline first.
-const WALL_CLOCK_TIMEOUT: Duration = Duration::from_secs(150);
-// Empty polls observed after producers stop before the partition is declared
drained.
-const DRAIN_EMPTY_POLLS: u32 = 20;
-
-/// Single-node, single shard (`"1"`) and multi shard (`"2"`).
-#[iggy_harness(
- cluster_nodes = 1,
- server(system.sharding.cpu_allocation = ["1", "2"])
-)]
-async fn
given_single_node_when_produce_consume_hammered_should_not_lose_messages(
- harness: &TestHarness,
-) {
- run_hammer(harness).await;
-}
-
-/// Three-node cluster, single shard (`"1"`) and multi shard (`"2"`) per node.
-/// Heavy (3 servers * N shards); run on demand with `--ignored`.
-#[iggy_harness(
- cluster_nodes = 3,
- server(system.sharding.cpu_allocation = ["1", "2"])
-)]
-#[ignore = "3-node cluster: heavy, run on demand with --ignored"]
-async fn given_cluster_when_produce_consume_hammered_should_not_lose_messages(
- harness: &TestHarness,
-) {
- run_hammer(harness).await;
-}
-
-async fn run_hammer(harness: &TestHarness) {
- tokio::time::timeout(WALL_CLOCK_TIMEOUT, run_hammer_inner(harness))
- .await
- .expect("stress test exceeded WALL_CLOCK_TIMEOUT; server likely
crashed at boot or a poll wedged");
-}
-
-async fn run_hammer_inner(harness: &TestHarness) {
- let stream_id = Identifier::named(STREAM_NAME).unwrap();
-
- let setup = harness.tcp_root_client().await.unwrap();
- setup.create_stream(STREAM_NAME).await.unwrap();
- setup
- .create_topic(
- &stream_id,
- TOPIC_NAME,
- 1,
- CompressionAlgorithm::None,
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .unwrap();
- drop(setup);
-
- let producer_done = Arc::new(AtomicBool::new(false));
-
- // Spawn consumers first so they poll concurrently with the producers from
- // the very first send. Each reads the whole partition independently by
- // explicit offset and asserts strict contiguity (no gap = no loss, no
dup).
- let mut consumers = Vec::with_capacity(CONSUMERS as usize);
- for consumer_id in 0..CONSUMERS {
- let client = harness.tcp_root_client().await.unwrap();
- let done = producer_done.clone();
- consumers.push(tokio::spawn(consume_partition(client, consumer_id,
done)));
- }
-
- // All producers hammer the single partition for HAMMER_DURATION.
- let mut producers = Vec::with_capacity(PRODUCERS as usize);
- for producer_id in 0..PRODUCERS {
- let client = harness.tcp_root_client().await.unwrap();
- producers.push(tokio::spawn(produce_partition(client, producer_id)));
- }
-
- // Producers stop at the hammer deadline; sum their sends, then signal
- // consumers to drain.
- let mut total_sent = 0u64;
- for handle in producers {
- total_sent += handle.await.unwrap();
- }
- producer_done.store(true, Ordering::Relaxed);
-
- assert!(
- total_sent > 0,
- "hammer produced no messages; workload wiring is broken"
- );
-
- // Every consumer independently read the full partition; each must have
seen
- // exactly the committed total, contiguously (asserted inside the task).
- for (consumer_id, handle) in consumers.into_iter().enumerate() {
- let received = handle.await.unwrap();
- assert_eq!(
- received, total_sent,
- "consumer {consumer_id}: consumed {received} != produced
{total_sent} (message loss)",
- );
- }
-
- let cleanup = harness.tcp_root_client().await.unwrap();
- cleanup.delete_stream(&stream_id).await.unwrap();
-}
-
-/// Send `PRODUCER_BATCH`-sized batches to the shared partition until the
hammer
-/// deadline. Returns the number of messages sent (each send awaits commit).
-async fn produce_partition(client: IggyClient, producer_id: u32) -> u64 {
- let stream = Identifier::named(STREAM_NAME).unwrap();
- let topic = Identifier::named(TOPIC_NAME).unwrap();
- let partitioning = Partitioning::partition_id(PARTITION_ID);
- let deadline = Instant::now() + HAMMER_DURATION;
- let mut sent = 0u64;
-
- while Instant::now() < deadline {
- let mut messages: Vec<IggyMessage> = (0..PRODUCER_BATCH)
- .map(|i| {
- IggyMessage::builder()
- .payload(Bytes::from(format!(
- "prod{producer_id}-{}",
- sent + u64::from(i)
- )))
- .build()
- .unwrap()
- })
- .collect();
- client
- .send_messages(&stream, &topic, &partitioning, &mut messages)
- .await
- .unwrap_or_else(|e| panic!("producer {producer_id} send failed at
sent={sent}: {e}"));
- sent += u64::from(PRODUCER_BATCH);
- }
- sent
-}
-
-/// Read the shared partition in full by explicit offset with `auto_commit =
-/// false`, asserting each message arrives at the next contiguous offset.
Drains
-/// until producers are done and `DRAIN_EMPTY_POLLS` consecutive empty polls
-/// confirm the tail. Returns the number of messages received.
-async fn consume_partition(
- client: IggyClient,
- consumer_id: u32,
- producer_done: Arc<AtomicBool>,
-) -> u64 {
- let stream = Identifier::named(STREAM_NAME).unwrap();
- let topic = Identifier::named(TOPIC_NAME).unwrap();
- let consumer = Consumer::default();
- let mut next_offset = 0u64;
- let mut received = 0u64;
- let mut consecutive_empty = 0u32;
- let deadline = Instant::now() + MAX_TEST_DURATION;
-
- loop {
- assert!(
- Instant::now() < deadline,
- "consumer {consumer_id} timed out: received {received},
next_offset {next_offset}"
- );
-
- let polled = match client
- .poll_messages(
- &stream,
- &topic,
- Some(PARTITION_ID),
- &consumer,
- &PollingStrategy::offset(next_offset),
- CONSUMER_BATCH,
- false,
- )
- .await
- {
- Ok(polled) => polled,
- Err(e) => {
- // Transient under load; back off and retry.
- eprintln!("consumer {consumer_id} poll error: {e:?}");
- tokio::time::sleep(Duration::from_millis(10)).await;
- continue;
- }
- };
-
- if polled.messages.is_empty() {
- if producer_done.load(Ordering::Relaxed) {
- consecutive_empty += 1;
- if consecutive_empty >= DRAIN_EMPTY_POLLS {
- break;
- }
- }
- tokio::time::sleep(Duration::from_millis(5)).await;
- continue;
- }
-
- consecutive_empty = 0;
- for msg in &polled.messages {
- assert_eq!(
- msg.header.offset, next_offset,
- "consumer {consumer_id} offset gap/dup: expected
{next_offset}, got {}",
- msg.header.offset
- );
- next_offset += 1;
- received += 1;
- }
- }
-
- received
-}
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 4ebe11638..148c4078b 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -6869,6 +6869,19 @@ mod purge_floor_tests {
"the fenced op must not linger in the staged-commit table"
);
+ // A store admitted after the purge carries a higher op -- the primary
+ // assigns them monotonically at admission -- so it lands above the
+ // floor and applies normally.
+ journal_store_offset(&mut partition, 2, 7, 4).await;
+ partition.consensus().advance_commit_max(2);
+ partition.commit_journal(&repair_config()).await;
+
+ assert_eq!(
+ partition.consumer_offsets.pin().len(),
+ 1,
+ "a store admitted after the purge must survive the floor"
+ );
+
let _ = std::fs::remove_dir_all(&dir);
}
