This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch replicated_crash_contract
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/replicated_crash_contract by
this push:
new c1bf37f16 address review
c1bf37f16 is described below
commit c1bf37f1683d188739bad34ffcd55318d7d1102b
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Wed Nov 4 16:01:02 2026 +0100
address review
---
core/integration/src/harness/disk.rs | 24 +++++
core/integration/tests/cluster/crash_durability.rs | 104 ++++++++++++---------
core/journal/src/partition_journal.rs | 46 ++++++---
3 files changed, 115 insertions(+), 59 deletions(-)
diff --git a/core/integration/src/harness/disk.rs
b/core/integration/src/harness/disk.rs
index 4762fd771..c45a233b2 100644
--- a/core/integration/src/harness/disk.rs
+++ b/core/integration/src/harness/disk.rs
@@ -27,6 +27,8 @@ use std::time::Duration;
use consensus::VsrState;
use iggy::prelude::{ClusterClient, ClusterNodeRole, ConsumerKind};
+use journal::PartitionPrepareJournal;
+use journal::partition_journal::FRONTIER_FILE_NAME;
use journal::superblock::{SLOT_FILE_NAMES, SuperblockContents, decode_slots};
use tokio::time::sleep;
@@ -43,6 +45,9 @@ const CONVERGENCE_STABLE_POLLS: u32 = 3;
const OFFSETS_DIR: &str = "offsets";
const CONSUMER_OFFSETS_DIR: &str = "consumers";
const GROUP_OFFSETS_DIR: &str = "groups";
+/// Prefix of a partition's WAL directory, `prepares-{created_revision}`, as
+/// `partitions::iggy_partition` names it.
+const PREPARES_DIR_PREFIX: &str = "prepares-";
/// Numeric regular offset files in one partition's selected kind directory.
pub fn consumer_offset_file_ids(
@@ -576,6 +581,25 @@ pub fn read_partition_superblock_state(data_path: &Path)
-> Option<VsrState> {
read_superblock_state_in(&dir, "partition")
}
+/// The message offset the single partition's published WAL frontier ends at on
+/// a node: a restart keeps the segment messages below it and discards the
rest.
+/// `None` while no frontier owning segment bodies exists. Read it while the
+/// node is down, since a live node keeps publishing. Intended for layouts with
+/// exactly one partition.
+pub fn read_partition_wal_frontier(data_path: &Path) -> Option<u64> {
+ let path = walk(data_path, &mut |path| {
+ path.file_name()
+ .is_some_and(|name| name == FRONTIER_FILE_NAME)
+ && path
+ .parent()
+ .and_then(Path::file_name)
+ .and_then(|name| name.to_str())
+ .is_some_and(|name| name.starts_with(PREPARES_DIR_PREFIX))
+ })?;
+ let frontier = fs::read(path).ok()?;
+ PartitionPrepareJournal::published_segment_tail(&frontier).map(|tail|
tail.next_offset)
+}
+
fn read_superblock_state_in(dir: &Path, plane: &str) -> Option<VsrState> {
let slot_a = fs::read(dir.join(SLOT_FILE_NAMES[0])).ok();
let slot_b = fs::read(dir.join(SLOT_FILE_NAMES[1])).ok();
diff --git a/core/integration/tests/cluster/crash_durability.rs
b/core/integration/tests/cluster/crash_durability.rs
index 7dacbb448..4698fcad4 100644
--- a/core/integration/tests/cluster/crash_durability.rs
+++ b/core/integration/tests/cluster/crash_durability.rs
@@ -62,28 +62,31 @@ const POLL_INTERVAL: Duration = Duration::from_millis(250);
/// primary dies, so the surviving quorum has a meaningful prefix to preserve.
const PRE_KILL_ACKS: usize = 50;
-async fn create_stream_and_topic(client: &IggyClient, eager_flush: bool) {
+async fn create_stream_and_topic(client: &IggyClient, eager_flush: bool,
durability: Durability) {
client
.create_stream(STREAM_NAME, &StreamCreateOptions::default())
.await
.expect("create stream");
- // `eager_flush` persists and fsyncs every committed batch on every
- // replica before anything else happens on the partition; it is a topic
- // creation option, so it travels with the topic to all nodes. The default
- // shape leaves the 1024-message / 1 MiB thresholds in place, so small
- // runs are acked from RAM only.
+ // `eager_flush` saves every committed batch to the segments on every
+ // replica before anything else happens on the partition. `persisted`
+ // durability also backs each ack with a stable-storage copy, where
+ // `replicated` acks on quorum commit alone. Both are topic creation
+ // options, so they travel with the topic to all nodes. The default shape
+ // leaves the 1024-message / 1 MiB thresholds in place, so small runs are
+ // acked from RAM only.
let options = if eager_flush {
TopicCreateOptions {
partitions_count: Some(1),
message_expiry: Some(IggyExpiry::NeverExpire),
messages_required_to_save: Some(1),
- durability: iggy_common::Durability::Persisted,
+ durability,
..TopicCreateOptions::default()
}
} else {
TopicCreateOptions {
partitions_count: Some(1),
message_expiry: Some(IggyExpiry::NeverExpire),
+ durability,
..TopicCreateOptions::default()
}
};
@@ -173,8 +176,8 @@ async fn produce_acked_when_serving(client: &IggyClient,
budget: Duration) -> (u
}
}
-/// The payloads of up to `count` messages from offset 0.
-async fn poll_payloads_from_start(client: &IggyClient, count: u32) ->
Vec<String> {
+/// The `(offset, payload)` of up to `count` messages from offset 0.
+async fn poll_from_start(client: &IggyClient, count: u32) -> Vec<(u64,
String)> {
let polled = client
.poll_messages(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -192,7 +195,12 @@ async fn poll_payloads_from_start(client: &IggyClient,
count: u32) -> Vec<String
polled
.messages
.iter()
- .map(|message| String::from_utf8_lossy(&message.payload).into_owned())
+ .map(|message| {
+ (
+ message.header.offset,
+ String::from_utf8_lossy(&message.payload).into_owned(),
+ )
+ })
.collect()
}
@@ -408,7 +416,7 @@ async fn
given_committed_sends_when_a_backup_is_killed_and_restarts_should_prese
harness: &mut TestHarness,
) {
let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, true).await;
+ create_stream_and_topic(&client, true, Durability::Persisted).await;
let mut acked = produce_acked(&client, "pre-kill", 200).await;
let leader = disk::leader_node_index(harness).await;
@@ -452,7 +460,7 @@ async fn
given_mid_flight_produce_when_the_primary_is_killed_should_elect_and_pr
harness: &mut TestHarness,
) {
let setup_client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&setup_client, false).await;
+ create_stream_and_topic(&setup_client, false,
Durability::Replicated).await;
let producer_client = harness.tcp_root_client().await.unwrap();
let stop = Arc::new(AtomicBool::new(false));
@@ -548,7 +556,7 @@ async fn
given_a_committed_consumer_offset_when_a_backup_is_killed_should_surviv
harness: &mut TestHarness,
) {
let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, false).await;
+ create_stream_and_topic(&client, false, Durability::Replicated).await;
let acked = produce_acked(&client, "offset-fodder", 10).await;
let stored_offset = acked.last().expect("ten acked sends").0;
@@ -591,7 +599,7 @@ async fn
given_eager_flush_topic_when_the_whole_cluster_is_killed_should_recover
harness: &mut TestHarness,
) {
let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, true).await;
+ create_stream_and_topic(&client, true, Durability::Persisted).await;
let acked = produce_acked(&client, "flushed", 30).await;
let payloads: Vec<String> = acked.iter().map(|(_, payload)|
payload.clone()).collect();
wait_until_payloads_installed(harness, &payloads,
FLUSH_INSTALL_TIMEOUT).await;
@@ -627,30 +635,33 @@ async fn
given_replicated_eager_flush_topic_when_the_whole_cluster_is_killed_sho
harness: &mut TestHarness,
) {
let client = harness.tcp_root_client().await.unwrap();
- client
- .create_stream(STREAM_NAME, &StreamCreateOptions::default())
- .await
- .expect("create stream");
- client
- .create_topic(
- &Identifier::named(STREAM_NAME).unwrap(),
- TOPIC_NAME,
- &TopicCreateOptions {
- partitions_count: Some(1),
- message_expiry: Some(IggyExpiry::NeverExpire),
- messages_required_to_save: Some(1),
- durability: Durability::Replicated,
- ..TopicCreateOptions::default()
- },
- )
- .await
- .expect("create topic");
+ create_stream_and_topic(&client, true, Durability::Replicated).await;
let acked = produce_acked(&client, "flushed", 30).await;
let payloads: Vec<String> = acked.iter().map(|(_, payload)|
payload.clone()).collect();
wait_until_payloads_installed(harness, &payloads,
FLUSH_INSTALL_TIMEOUT).await;
// Same reason as above for keeping the producer's connection open.
harness.kill_cluster().expect("SIGKILL every node");
+ // Read with every node down, so no publication can land after the read.
+ let frontiers: Vec<Option<u64>> = (0..harness.cluster_size())
+ .map(|node|
disk::read_partition_wal_frontier(&harness.node(node).data_path()))
+ .collect();
+ // Whichever log the election picks, recovery then keeps the same prefix.
+ let frontier = frontiers[0].expect("node 0 published a partition WAL
frontier");
+ assert!(
+ frontiers.iter().all(|other| *other == Some(frontier)),
+ "every node must publish the same WAL frontier: {frontiers:?}"
+ );
+ let (mut expected, discarded): (Vec<_>, Vec<_>) = acked
+ .iter()
+ .cloned()
+ .partition(|(offset, _)| *offset < frontier);
+ assert!(
+ !discarded.is_empty(),
+ "the published WAL frontier {frontier} covers all {} acks, so the
crash pins no loss",
+ acked.len()
+ );
+
harness
.restart_cluster()
.await
@@ -661,18 +672,21 @@ async fn
given_replicated_eager_flush_topic_when_the_whole_cluster_is_killed_sho
// A write that reads back proves the partition serves again, so the
// missing acks below are discarded, not still recovering.
let after_restart = produce_acked_when_serving(&client,
CONVERGE_TIMEOUT).await;
- wait_for_acked_readable(&client, &[after_restart], CONVERGE_TIMEOUT)
- .await
- .unwrap_or_else(|state| panic!("the partition should serve after the
restart: {state}"));
- let readable = poll_payloads_from_start(&client, acked.len() as u32 +
16).await;
- let survived = payloads
- .iter()
- .filter(|payload| readable.contains(payload))
- .count();
- assert!(
- survived < acked.len(),
- "a whole-cluster SIGKILL discards replicated acks past the published
WAL frontier, but \
- all {survived} survived"
+ wait_for_acked_readable(
+ &client,
+ std::slice::from_ref(&after_restart),
+ CONVERGE_TIMEOUT,
+ )
+ .await
+ .unwrap_or_else(|state| panic!("the partition should serve after the
restart: {state}"));
+ expected.push(after_restart);
+ let readable = poll_from_start(&client, acked.len() as u32 + 16).await;
+ assert_eq!(
+ readable,
+ expected,
+ "a whole-cluster SIGKILL must keep exactly the acks below the
published WAL frontier \
+ {frontier} and discard the {} past it",
+ discarded.len()
);
}
@@ -1015,7 +1029,7 @@ async fn
given_uninitialized_partitions_when_all_replicas_restart_should_elect_a
harness: &mut TestHarness,
) {
let client = harness.tcp_root_client().await.unwrap();
- create_stream_and_topic(&client, true).await;
+ create_stream_and_topic(&client, true, Durability::Persisted).await;
let stream = Identifier::named(STREAM_NAME).unwrap();
let topic = Identifier::named(TOPIC_NAME).unwrap();
let details = client.get_topic(&stream, &topic).await.unwrap().unwrap();
diff --git a/core/journal/src/partition_journal.rs
b/core/journal/src/partition_journal.rs
index 536b430a2..342a7c0a9 100644
--- a/core/journal/src/partition_journal.rs
+++ b/core/journal/src/partition_journal.rs
@@ -168,6 +168,18 @@ impl PartitionPrepareJournal {
Ok(state.is_some())
}
+ /// The segment tail in whichever slot of `frontier`, the bytes of a
+ /// [`FRONTIER_FILE_NAME`] file, recovery would select. When every slot
+ /// verifies, recovery truncates the tail segment to it. `None` when no
slot
+ /// verifies or the WAL does not own segment bodies.
+ #[must_use]
+ pub fn published_segment_tail(frontier: &[u8]) -> Option<SegmentPosition> {
+ let (state, _, _) = newest_frontier_slot(frontier);
+ state?
+ .segment_storage
+ .map(|segments| segments.tail.position)
+ }
+
/// # Errors
/// Returns an error on I/O failure or invalid durable history.
pub async fn open(directory: &Path, group: u64, incarnation: u64) ->
io::Result<Self> {
@@ -402,23 +414,11 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
file: Option<&S::File>,
slots: usize,
) -> io::Result<(Option<JournalState>, u64, usize)> {
- let mut newest = None;
- let mut sequence = 0;
- let mut verified = 0;
let Some(file) = file.filter(|_| slots > 0) else {
- return Ok((newest, sequence, verified));
+ return Ok((None, 0, 0));
};
let bytes = file.read(0, slots * PARTITION_WAL_BLOCK_SIZE).await?;
- for slot in bytes.as_chunks::<PARTITION_WAL_BLOCK_SIZE>().0 {
- if let Ok((state, slot_sequence)) = JournalState::decode(slot) {
- verified += 1;
- if newest.is_none() || slot_sequence > sequence {
- newest = Some(state);
- sequence = slot_sequence;
- }
- }
- }
- Ok((newest, sequence, verified))
+ Ok(newest_frontier_slot(&bytes))
}
pub const fn certified_log_view(&self) -> Option<u32> {
@@ -1650,6 +1650,24 @@ const fn frontier_offset(sequence: u64) -> u64 {
(sequence % FRONTIER_SLOTS as u64) * PARTITION_WAL_BLOCK_SIZE as u64
}
+/// The newest verifying slot in `bytes`, its publication sequence, and how
+/// many whole slots verified. See [`PartitionPrepareJournal::read_frontier`].
+fn newest_frontier_slot(bytes: &[u8]) -> (Option<JournalState>, u64, usize) {
+ let mut newest = None;
+ let mut sequence = 0;
+ let mut verified = 0;
+ for slot in bytes.as_chunks::<PARTITION_WAL_BLOCK_SIZE>().0 {
+ if let Ok((state, slot_sequence)) = JournalState::decode(slot) {
+ verified += 1;
+ if newest.is_none() || slot_sequence > sequence {
+ newest = Some(state);
+ sequence = slot_sequence;
+ }
+ }
+ }
+ (newest, sequence, verified)
+}
+
/// Padded size of a prepare record, including its envelope.
///
/// # Errors
