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/10/23 07:25:13 UTC
(iggy) branch master updated: test(integration): add primary-side torn-tail crash recovery specs (#4442)
This is an automated email from the ASF dual-hosted git repository.
numinnex 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 2b4745219 test(integration): add primary-side torn-tail crash recovery specs (#4442)
2b4745219 is described below
commit 2b47452196e5d6abec9cdb947781a0e2605c59b8
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Fri Oct 23 09:25:07 2026 +0200
test(integration): add primary-side torn-tail crash recovery specs (#4442)
Closes #4441
---
.../tests/cluster/crash_recovery_corruption.rs | 230 ++++++++++++++-------
1 file changed, 161 insertions(+), 69 deletions(-)
diff --git a/core/integration/tests/cluster/crash_recovery_corruption.rs b/core/integration/tests/cluster/crash_recovery_corruption.rs
index 339b03a2d..f892dbace 100644
--- a/core/integration/tests/cluster/crash_recovery_corruption.rs
+++ b/core/integration/tests/cluster/crash_recovery_corruption.rs
@@ -65,6 +65,8 @@ const GARBAGE_BYTE: u8 = 0xA5;
/// Metadata ops produced before the WAL surgery, so a flip at one quarter of
/// the file provably lands ahead of many complete committed entries.
const WAL_FODDER_STREAMS: usize = 20;
+/// View of a partition group that never left view 0.
+const INITIAL_VIEW: u32 = 0;
/// Sparse index entry layout, mirrored from `partitions::iggy_index::IggyIndex`
/// (which the `integration` crate does not depend on): three little-endian
@@ -119,6 +121,33 @@ async fn create_stream_and_topic(client: &IggyClient, durability: Durability) {
.expect("create topic");
}
+/// Create the test topic, produce `count` batches through the harness root
+/// client and wait until every node's segment files hold them, so surgery
+/// always finds the payloads on disk. The setup client is dropped before
+/// returning.
+async fn seed_flushed_batches(
+ harness: &TestHarness,
+ durability: Durability,
+ payload_prefix: &str,
+ count: u32,
+) -> Vec<(u64, String)> {
+ let client = harness.tcp_root_client().await.unwrap();
+ create_stream_and_topic(&client, durability).await;
+ let acked = produce_acked(&client, payload_prefix, count).await;
+ let payloads: Vec<String> = acked.iter().map(|(_, payload)| payload.clone()).collect();
+ for node in 0..harness.cluster_size() {
+ wait_until_node_holds_payloads(
+ harness,
+ node,
+ &payloads,
+ FLUSH_INSTALL_TIMEOUT,
+ "eager flush before the surgery",
+ )
+ .await;
+ }
+ acked
+}
+
/// Send `count` single-message batches, returning each confirmed
/// `(base_offset, payload)`.
async fn produce_acked(
@@ -292,6 +321,46 @@ async fn pick_backup(harness: &TestHarness) -> (usize, usize) {
(leader, backup)
}
+/// Primary of the highest partition view any of `nodes` has persisted. The
+/// metadata and partition planes change views independently, so the roster
+/// leader does not name the partition primary.
+fn partition_primary(harness: &TestHarness, nodes: &[usize]) -> usize {
+ let view = nodes
+ .iter()
+ .map(|&node| {
+ disk::read_partition_superblock_state(&harness.node(node).data_path())
+ .map_or(INITIAL_VIEW, |state| state.view)
+ })
+ .max()
+ .unwrap_or(INITIAL_VIEW);
+ view as usize % harness.cluster_size()
+}
+
+/// Poll the survivors' partition superblocks until a view whose primary is not
+/// `former` is persisted; until then `former` still owns the partition.
+async fn wait_for_partition_successor(
+ harness: &TestHarness,
+ former: usize,
+ budget: Duration,
+) -> usize {
+ let survivors: Vec<usize> = (0..harness.cluster_size())
+ .filter(|index| *index != former)
+ .collect();
+ let deadline = tokio::time::Instant::now() + budget;
+ loop {
+ let primary = partition_primary(harness, &survivors);
+ if primary != former {
+ return primary;
+ }
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "the survivors did not elect a partition primary to succeed node {former} within \
+ {budget:?}"
+ );
+ sleep(POLL_INTERVAL).await;
+ }
+}
+
/// The ACTIVE (highest base offset) partition segment file with `extension`
/// under a node's data dir.
fn find_active_segment_file(data_path: &Path, extension: &str) -> PathBuf {
@@ -594,32 +663,15 @@ async fn given_index_entries_past_the_checkpoint_when_a_killed_node_reboots_shou
/// node survives its NEXT restart (a stale cursor bricked it as message/index
/// divergence), every acked offset still reads back, and the at-rest `.log`
/// bytes stay identical across replicas (resurrected garbage diverged them).
-// TODO(hubcio): both torn-tail specs tear a BACKUP; add a primary-side
-// variant (tear the leader's segment, restart it) since the leader path
-// exercises different reopen and catch-up code.
#[iggy_harness(cluster_nodes = 3)]
async fn given_a_torn_segment_tail_when_a_node_recovers_should_keep_size_counter_consistent(
harness: &mut TestHarness,
) {
- let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, Durability::Persisted).await;
- let mut acked = produce_acked(&client, "pre-torn", 30).await;
- let pre_payloads: Vec<String> = acked.iter().map(|(_, payload)| payload.clone()).collect();
- for node in 0..harness.cluster_size() {
- wait_until_node_holds_payloads(
- harness,
- node,
- &pre_payloads,
- FLUSH_INSTALL_TIMEOUT,
- "eager flush before the surgery",
- )
- .await;
- }
+ let mut acked = seed_flushed_batches(harness, Durability::Persisted, "pre-torn", 30).await;
- // The setup client is pinned to node 0, which is the backup whenever the
- // leader sits elsewhere; harness clients never reconnect, so swap to a
- // producer on the leader, which stays up across both backup restarts.
- drop(client);
+ // The setup client was pinned to node 0, which is the backup whenever the
+ // leader sits elsewhere; harness clients never reconnect, so produce
+ // through the leader, which stays up across both backup restarts.
let (leader, backup) = pick_backup(harness).await;
let client = harness
.root_client_for_node(leader)
@@ -687,25 +739,12 @@ async fn given_a_torn_segment_tail_when_a_node_recovers_should_keep_size_counter
async fn given_a_torn_index_tail_when_a_node_recovers_should_not_misalign_subsequent_entries(
harness: &mut TestHarness,
) {
- let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, Durability::Persisted).await;
- let mut acked = produce_acked(&client, "pre-torn-index", 30).await;
- let pre_payloads: Vec<String> = acked.iter().map(|(_, payload)| payload.clone()).collect();
- for node in 0..harness.cluster_size() {
- wait_until_node_holds_payloads(
- harness,
- node,
- &pre_payloads,
- FLUSH_INSTALL_TIMEOUT,
- "eager flush before the surgery",
- )
- .await;
- }
+ let mut acked =
+ seed_flushed_batches(harness, Durability::Persisted, "pre-torn-index", 30).await;
- // The setup client is pinned to node 0, which is the backup whenever the
- // leader sits elsewhere; harness clients never reconnect, so swap to a
- // producer on the leader, which stays up across both backup restarts.
- drop(client);
+ // The setup client was pinned to node 0, which is the backup whenever the
+ // leader sits elsewhere; harness clients never reconnect, so produce
+ // through the leader, which stays up across both backup restarts.
let (leader, backup) = pick_backup(harness).await;
let client = harness
.root_client_for_node(leader)
@@ -746,6 +785,87 @@ async fn given_a_torn_index_tail_when_a_node_recovers_should_not_misalign_subseq
});
}
+/// Primary-side counterpart of the backup torn-tail specs: the torn file is
+/// the partition primary's. A restarted node rejoins as a backup under an
+/// elected successor, so the post-torn batches reach its reopened segment by
+/// replication. Same contract as the backup `.log` spec: boot over the tear,
+/// survive the NEXT restart after that append, every acked offset reads back,
+/// at-rest segment bytes identical.
+async fn require_torn_primary_recovery(harness: &mut TestHarness, extension: &str, garbage: usize) {
+ let mut acked = seed_flushed_batches(harness, Durability::Persisted, "pre-torn", 30).await;
+
+ let nodes: Vec<usize> = (0..harness.cluster_size()).collect();
+ let primary = partition_primary(harness, &nodes);
+ harness.stop_node(primary).expect("stop the primary");
+ let torn_file = find_active_segment_file(&harness.node(primary).data_path(), extension);
+ append_garbage(&torn_file, garbage);
+
+ harness.restart_node(primary).unwrap_or_else(|error| {
+ panic!("a primary must boot over a torn segment .{extension} tail: {error}")
+ });
+
+ let successor = wait_for_partition_successor(harness, primary, CONVERGE_TIMEOUT).await;
+ let client = harness
+ .root_client_for_node(successor)
+ .await
+ .expect("connect a producer to the successor primary");
+ acked.extend(produce_acked(&client, "post-torn", 30).await);
+ let all_payloads: Vec<String> = acked.iter().map(|(_, payload)| payload.clone()).collect();
+ for node in 0..harness.cluster_size() {
+ wait_until_node_holds_payloads(
+ harness,
+ node,
+ &all_payloads,
+ CONVERGE_TIMEOUT,
+ "replication onto the reopened segment after the torn-tail recovery",
+ )
+ .await;
+ }
+
+ harness.restart_node(primary).unwrap_or_else(|error| {
+ panic!(
+ "a node must restart cleanly over a segment it repaired as the primary and then \
+ appended to as a backup; boot error: {error}"
+ )
+ });
+ let client = wait_until_cluster_serves(harness, &nodes, CONVERGE_TIMEOUT).await;
+ wait_for_acked_readable(&client, &acked, CONVERGE_TIMEOUT)
+ .await
+ .unwrap_or_else(|state| {
+ panic!(
+ "every acked offset must poll back after the primary-side torn-tail recovery: \
+ {state}"
+ )
+ });
+
+ let data_paths: Vec<PathBuf> = harness
+ .all_servers()
+ .iter()
+ .map(|server| server.data_path())
+ .collect();
+ harness
+ .stop()
+ .await
+ .expect("stop the cluster for the at-rest comparison");
+ disk::assert_replica_data_identical(&data_paths, false);
+}
+
+/// Primary-side `.log` variant; see `TORN_LOG_GARBAGE` for the torn shape.
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_torn_segment_tail_when_the_primary_recovers_should_rejoin_without_diverging(
+ harness: &mut TestHarness,
+) {
+ require_torn_primary_recovery(harness, "log", TORN_LOG_GARBAGE).await;
+}
+
+/// Primary-side `.index` variant; see `TORN_INDEX_GARBAGE` for the torn shape.
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_torn_index_tail_when_the_primary_recovers_should_rejoin_without_diverging(
+ harness: &mut TestHarness,
+) {
+ require_torn_primary_recovery(harness, "index", TORN_INDEX_GARBAGE).await;
+}
+
/// Replicated storage rebuilds a stale index from the surviving log and refills
/// the tail from peers. Persisted storage syncs bodies before publishing WAL
/// references, so deleting that tail must quarantine the damaged partition.
@@ -756,22 +876,8 @@ async fn given_an_index_ahead_of_a_truncated_log_when_a_node_recovers_should_pre
harness: &mut TestHarness,
durability: Durability,
) {
- let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, durability).await;
- let acked = produce_acked(&client, "index-ahead", INDEX_AHEAD_BATCHES).await;
+ let acked = seed_flushed_batches(harness, durability, "index-ahead", INDEX_AHEAD_BATCHES).await;
let payloads: Vec<String> = acked.iter().map(|(_, payload)| payload.clone()).collect();
- for node in 0..harness.cluster_size() {
- wait_until_node_holds_payloads(
- harness,
- node,
- &payloads,
- FLUSH_INSTALL_TIMEOUT,
- "eager flush before the surgery",
- )
- .await;
- }
-
- drop(client);
// The leader keeps quorum and serving throughout, so the damaged node
// always has a peer to catch up from. SIGKILL rather than a graceful stop:
@@ -1037,21 +1143,7 @@ async fn given_interior_wal_corruption_when_a_node_boots_should_refuse_and_heal_
async fn given_a_corrupt_superblock_slot_when_a_node_boots_should_refuse_boot(
harness: &mut TestHarness,
) {
- let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, Durability::Persisted).await;
- let mut acked = produce_acked(&client, "pre-fault", 20).await;
- let pre_payloads: Vec<String> = acked.iter().map(|(_, payload)| payload.clone()).collect();
- for node in 0..harness.cluster_size() {
- wait_until_node_holds_payloads(
- harness,
- node,
- &pre_payloads,
- FLUSH_INSTALL_TIMEOUT,
- "eager flush before the surgery",
- )
- .await;
- }
- drop(client);
+ let mut acked = seed_flushed_batches(harness, Durability::Persisted, "pre-fault", 20).await;
// A quiet cluster at view 0 has an EMPTY metadata superblock: the record
// is first persisted when a replica adopts an advanced view. Force a view