You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/07 10:00:39 UTC
(iggy) branch fix_integration_flakes created (now faf8d7a30)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a change to branch fix_integration_flakes
in repository https://gitbox.apache.org/repos/asf/iggy.git
at faf8d7a30 test(integration): fix vsr flakes from racing recovery routes
This branch includes the following new commits:
new faf8d7a30 test(integration): fix vsr flakes from racing recovery routes
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: test(integration): fix vsr flakes from racing recovery routes
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch fix_integration_flakes
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit faf8d7a30e8b834e1201110eb325e9df4480d4d0
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Fri Aug 7 12:00:28 2026 +0200
test(integration): fix vsr flakes from racing recovery routes
---
.../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 | 126 ++++++++++
9 files changed, 339 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 dad3f70e3..54a31ee67 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -158,6 +158,24 @@ where
/// generation against this and resets only when it advances, so a redundant
/// reconcile pass never re-wipes a partition already at this generation.
pub(crate) applied_purge_generation: u64,
+ /// Highest partition op the last local purge superseded. A consumer-offset
+ /// op journaled before the purge but committed after it must NOT be
+ /// applied: `PurgeTopic` rides the metadata plane while
+ /// `StoreConsumerOffset*` rides this one, so the two logs carry no mutual
+ /// ordering, and a server-side auto-commit is dispatched with no client
+ /// waiting on it (`server-ng::dispatch::submit_auto_commit`) -- it can
+ /// still be in flight when the purge lands. Without this fence the commit
+ /// walk rewrites the offset map AND the offset file after the purge
+ /// cleared both, and nothing ever clears them again: this replica's
+ /// `applied_purge_generation` already matches the committed one, so the
+ /// reconciler sees no outstanding purge. The purge already fences the
+ /// other asynchronous writer of this state (the state-transfer install);
+ /// this is the symmetric fence for the offset-commit path.
+ ///
+ /// Memory-only on purpose, exactly like `applied_purge_generation`: a
+ /// restarted replica rehydrates its offsets from an already-empty
+ /// directory, so there is nothing for a persisted fence to protect.
+ pub(crate) consumer_offset_purge_fence: u64,
/// Durable superblock for this partition's consensus group, recording
/// `(view, log_view)` across a crash so this replica can never
/// re-participate in a view older than one it advertised. `None` for
@@ -268,6 +286,10 @@ where
.field("recovered_durable_offset", &self.recovered_durable_offset)
.field("observed_view", &self.observed_view)
.field("applied_purge_generation", &self.applied_purge_generation)
+ .field(
+ "consumer_offset_purge_fence",
+ &self.consumer_offset_purge_fence,
+ )
.finish_non_exhaustive()
}
}
@@ -436,6 +458,7 @@ where
persisted_offsets: RefCell::new(HashMap::new()),
observed_view,
applied_purge_generation: 0,
+ consumer_offset_purge_fence: 0,
superblock: None,
superblock_lock: LocalGate::new(),
superblock_write_failures: Cell::new(0),
@@ -964,6 +987,17 @@ where
&mut self,
op: u64,
) -> Result<(), IggyError> {
+ // Superseded by a purge: this op was journaled before the partition was
+ // reset, so its offset describes messages that no longer exist. Drop it
+ // rather than let the commit walk write it back over the cleared state
+ // (see `consumer_offset_purge_fence`). A store admitted after the purge
+ // carries a higher op -- the primary assigns them monotonically at
+ // admission -- so it lands above the fence and applies normally, and a
+ // suppressed in-flight auto-commit is re-derived from the next poll.
+ if op <= self.consumer_offset_purge_fence {
+ self.pending_consumer_offset_commits.remove(&op);
+ return Ok(());
+ }
// Peek (copy) instead of remove: if `persist_consumer_offset_commit`
// fails (e.g. disk full, fd exhausted) the pending entry must remain
// stageable for retry on the next apply. Removing first would strand
@@ -3823,6 +3857,16 @@ where
self.installed_frontier = None;
self.segment_checksum_cache.borrow_mut().clear();
+ // Fence the offset-commit path before clearing anything: every op the
+ // journal already holds predates this purge, so committing one later
+ // must not resurrect the offset the clear below is about to drop. The
+ // sequencer's current value is the right line -- it covers ops that are
+ // journaled but not yet committed, which is exactly where an in-flight
+ // auto-commit sits. Staged entries for those ops go too, so the commit
+ // walk cannot apply one straight out of the staging table.
+ self.consumer_offset_purge_fence = self.consensus().sequencer().current_sequence();
+ self.pending_consumer_offset_commits.clear();
+
// Clear consumer + consumer-group offsets (memory + disk). Collect the
// file paths before deleting so the map guard is not held across an
// await.
@@ -5714,6 +5758,88 @@ mod tests {
let _ = std::fs::remove_dir_all(&dir);
}
+ /// A consumer-offset op journaled before a purge must not resurrect the
+ /// offset when its commit lands afterwards. `PurgeTopic` rides the metadata
+ /// plane while `StoreConsumerOffset*` rides this one, so the two logs carry
+ /// no mutual ordering; a server auto-commit in particular is dispatched
+ /// with nobody waiting on it and can still be in flight when the purge
+ /// runs. Applying it afterwards used to rewrite both the offset map and the
+ /// offset file over the cleared state, permanently: this replica's
+ /// `applied_purge_generation` already matches the committed one, so the
+ /// reconciler sees no outstanding purge to re-run.
+ #[compio::test]
+ async fn purge_fences_a_consumer_offset_op_journaled_before_it() {
+ let dir = std::env::temp_dir().join(format!(
+ "iggy-purge-offset-fence-{}-{}",
+ std::process::id(),
+ std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .expect("system clock after epoch")
+ .as_nanos(),
+ ));
+ compio::fs::create_dir_all(&dir)
+ .await
+ .expect("create temp partition dir");
+
+ let mut partition = test_partition();
+ partition.set_partition_dir(dir.to_string_lossy().into_owned());
+ let consumer_id = 7u32;
+
+ // Journalled (staged) before the purge, not yet committed.
+ partition.consensus().sequencer().set_sequence(9);
+ partition.stage_consumer_offset_upsert(
+ 9,
+ ConsumerKind::ConsumerGroup,
+ consumer_id,
+ 9,
+ true,
+ );
+
+ partition
+ .purge(&repair_config(), 1)
+ .await
+ .expect("purge partition");
+
+ assert_eq!(
+ partition.consumer_offset_purge_fence, 9,
+ "the purge must fence every op the journal already held",
+ );
+ assert!(
+ partition.pending_consumer_offset_commits.is_empty(),
+ "the purge must drop staged pre-purge offset commits",
+ );
+
+ // The commit walk reaching that op afterwards is a no-op, not an error:
+ // refusing it would fail the commit and wedge the partition.
+ partition
+ .apply_staged_consumer_offset_commit(9)
+ .await
+ .expect("a fenced commit must not fail the commit walk");
+ assert!(
+ partition.consumer_group_offsets.pin().is_empty(),
+ "a pre-purge offset must not be resurrected after the purge",
+ );
+
+ // A store admitted after the purge carries a higher op -- the primary
+ // assigns them monotonically at admission -- so it still applies.
+ partition.stage_consumer_offset_upsert(
+ 10,
+ ConsumerKind::ConsumerGroup,
+ consumer_id,
+ 4,
+ false,
+ );
+ partition
+ .apply_staged_consumer_offset_commit(10)
+ .await
+ .expect("a post-purge commit applies");
+ assert_eq!(
+ partition.consumer_group_offsets.pin().len(),
+ 1,
+ "a store admitted after the purge must survive the fence",
+ );
+ }
+
/// Purge unlinks every segment and recreates the same paths, so a poll
/// suspended across the purge must not keep serving the old inodes through
/// its cached read state. The wipe reaches the in-flight clone through the