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
commit 0fd97a3eec6b598aab0691170569d70cf0cc264a Author: Grzegorz Koszyk <[email protected]> AuthorDate: Tue Nov 3 17:18:57 2026 +0100 test(integration): pin the crash loss of replicated partitions --- README.md | 2 +- core/integration/tests/cluster/crash_durability.rs | 131 ++++++++++++++++++++- core/server/config.toml | 1 + 3 files changed, 129 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index e21f035e7..ece44b5d2 100644 --- a/README.md +++ b/README.md @@ -289,7 +289,7 @@ What each policy survives in a cluster of `n` replicas. A partition gets the `pe - `persisted` keeps every acknowledged write through any pattern of crashes and restarts. It is needed to survive `q` or more replicas crashing at once, power loss of the whole cluster, or any crash in a one-replica cluster. - `replicated` keeps every acknowledged write as long as fewer than `q` replicas lose unflushed writes at the same time, where `q`, the replicas that acknowledge each write, is half the replicas rounded up but at most 3 (2 in a two-replica cluster). A replica that returns with that partition awaiting a state transfer counts as one of them: its log is not used, even if its WAL still holds ops. -- Graceful restarts flush the WAL first and lose nothing, one at a time or all together. A crash (SIGKILL, OOM kill) loses the WAL writes not yet handed to the OS, and power loss also loses what the OS had not written to disk. Under `replicated`, `q` or more such crashes at once can lose the most recently acknowledged writes. +- Graceful restarts flush the WAL first and lose nothing, one at a time or all together. After a crash (SIGKILL, OOM kill) or power loss, a replica recovers a `replicated` partition up to the last WAL frontier it published, at a checkpoint, a durable operation or a graceful shutdown, and discards later writes even if the OS held them. Under `replicated`, `q` or more such crashes at once can lose the most recently acknowledged writes. In a one-replica cluster, any crash can. - A restarted `replicated` replica accepts no writes and acknowledges no prepares until it joins a settled view: one a running primary sends it, or one that an election starts. Until then writes and polls sent to it, binary or HTTP, fail with `TransientNotAccepted` (HTTP 503), which clients retry. When no primary answers, the restarted replicas hold an election and mark their logs as possibly incomplete. A new view starts only from `n - q + 1` replicas whose logs are not marked (2 in a two-replica cluster, a majority up to six replicas), or once every replica has reported, and the partition stays unavailable until then. After a full-cluster restart that means every replica must be back. A replica whose partition awaits a state transfer still reports without a log while it is recovering; once it has joined a view it stays out of later elections until its transfer completes. A replica that never returns halts the partition indefinitely (no operator override yet). diff --git a/core/integration/tests/cluster/crash_durability.rs b/core/integration/tests/cluster/crash_durability.rs index 57a56d7e0..7dacbb448 100644 --- a/core/integration/tests/cluster/crash_durability.rs +++ b/core/integration/tests/cluster/crash_durability.rs @@ -24,11 +24,13 @@ //! quorum COMMIT, which for a topic below its flush thresholds means the //! batch lives only in the in-memory partition journal //! (`PartitionJournalMemStorage`) on each node. Process death never loses a -//! completed `write()` (the page cache survives the process), so what SIGKILL -//! probes is precisely that written-before-acked gap, not fsync ordering. +//! completed `write()` (the page cache survives the process), but recovery +//! only trusts what the published WAL frontier covers, and discards the bytes +//! of a `replicated` partition past it even when they reached the segments. //! -//! Single-node-crash tests must hold every ack via the surviving quorum; -//! the whole-cluster-crash test must hold exactly what reached the segments. +//! Single-node-crash tests must hold every ack via the surviving quorum. The +//! whole-cluster-crash tests must hold what was made durable, and pin that a +//! `replicated` partition loses the acks past its published frontier. use std::collections::HashMap; use std::path::PathBuf; @@ -132,6 +134,68 @@ async fn produce_acked( acked } +/// Send one message once the partition accepts writes again after a restart, +/// returning its confirmed `(offset, payload)`. +async fn produce_acked_when_serving(client: &IggyClient, budget: Duration) -> (u64, String) { + let stream = Identifier::named(STREAM_NAME).unwrap(); + let topic = Identifier::named(TOPIC_NAME).unwrap(); + let payload = "after-restart".to_owned(); + let deadline = tokio::time::Instant::now() + budget; + loop { + let mut messages = vec![ + IggyMessage::builder() + .payload(payload.clone().into()) + .build() + .expect("build message"), + ]; + match client + .send_messages( + &stream, + &topic, + &Partitioning::partition_id(PARTITION_ID), + &mut messages, + ) + .await + { + Ok(response) => { + let confirmation = response + .confirmations + .first() + .expect("the VSR server confirms every send"); + return (confirmation.base_offset, payload); + } + Err(error) => assert!( + tokio::time::Instant::now() < deadline, + "the partition did not accept a write after the restart: {error}" + ), + } + sleep(POLL_INTERVAL).await; + } +} + +/// The payloads of up to `count` messages from offset 0. +async fn poll_payloads_from_start(client: &IggyClient, count: u32) -> Vec<String> { + let polled = client + .poll_messages( + &Identifier::named(STREAM_NAME).unwrap(), + &Identifier::named(TOPIC_NAME).unwrap(), + PARTITION_ID, + &Consumer::default(), + &PollingStrategy::offset(0), + PollOptions { + count, + ..PollOptions::default() + }, + ) + .await + .expect("poll from offset 0"); + polled + .messages + .iter() + .map(|message| String::from_utf8_lossy(&message.payload).into_owned()) + .collect() +} + /// Poll from offset 0 until every acked `(offset, payload)` reads back at its /// confirmed offset with offsets strictly increasing, or `budget` runs out. /// `Err` carries the last observed shortfall so the caller's verdict can show @@ -553,6 +617,65 @@ 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(); + 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"); + 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"); + 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, &[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" + ); +} + /// The kill follows the confirmation without waiting for segment installation. /// This exercises the prepare WAL rather than graceful shutdown or page-cache flushes. #[iggy_harness(cluster_nodes = 3)] diff --git a/core/server/config.toml b/core/server/config.toml index ac9dfb700..7d94aa7e0 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -1002,6 +1002,7 @@ change_log_activity_capacity = 16384 # Persistence barriers can also persist co-batched messages. # Persisted prepares are limited to 64 MiB including the consensus header. Oversized operations are rejected permanently. # Both policies write data to storage. "replicated" waits for VSR commit and application without an additional stable-storage barrier. +# After a crash, a "replicated" partition recovers up to the last WAL frontier it published: at checkpoints, durable operations and graceful shutdown. # "persisted" also requires recoverable stable-storage copies at the replication quorum before explicit success. # A singleton quorum is one replica. Its persisted completion follows recoverable local persistence. # Poll auto-commit remains asynchronous. A successful poll does not confirm that its submitted offset operation committed or became durable.
