hubcio commented on code in PR #4738:
URL: https://github.com/apache/iggy/pull/4738#discussion_r4474566703
##########
core/integration/tests/cluster/crash_durability.rs:
##########
@@ -60,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 {
Review Comment:
simplification: both branches repeat the same topic options except for the
flush threshold. use one initializer with `messages_required_to_save:
eager_flush.then_some(1)`.
##########
core/integration/tests/cluster/crash_durability.rs:
##########
@@ -553,6 +625,71 @@ async fn
given_eager_flush_topic_when_the_whole_cluster_is_killed_should_recover
});
}
+/// The `replicated` twin of the eager-flush test above. A crash rolls a
+/// `replicated` partition back to the last WAL frontier it published, at a
+/// checkpoint, a durable operation or a graceful shutdown, so acks past it are
+/// lost even with every payload already in every node's segment files. Topics
+/// that must survive a whole-cluster crash use `persisted`.
+#[iggy_harness(cluster_nodes = 3)]
+async fn
given_replicated_eager_flush_topic_when_the_whole_cluster_is_killed_should_discard_acks_past_the_published_frontier(
+ harness: &mut TestHarness,
+) {
+ let client = harness.tcp_root_client().await.unwrap();
+ 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
+ .expect("restart the whole cluster");
+
+ let nodes: Vec<usize> = (0..harness.cluster_size()).collect();
+ let client = wait_until_cluster_serves(harness, &nodes,
CONVERGE_TIMEOUT).await;
+ // 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,
+ std::slice::from_ref(&after_restart),
Review Comment:
warning: the restart probe reads only 17 messages from offset zero, so
correct recovery of 17-29 earlier messages hides the new message and causes a
timeout. append `after_restart` to `expected` first, then pass `&expected` to
`wait_for_acked_readable`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
