This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch fix_viewchange_untrusted_logs in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e72fe9385b07f94b09ceefca6252d5f7cadddfaf Author: Grzegorz Koszyk <[email protected]> AuthorDate: Mon Oct 12 17:06:52 2026 +0200 temp --- README.md | 11 + core/binary_protocol/src/consensus/header.rs | 93 +++- core/consensus/src/dvc_merge.rs | 292 +++++++++- core/consensus/src/impls.rs | 800 ++++++++++++++++++++++++++- core/consensus/src/observability.rs | 16 + core/consensus/src/plane_helpers.rs | 9 +- core/consensus/src/view_change_quorum.rs | 3 + core/metadata/src/stm/stream.rs | 34 +- core/partitions/src/iggy_partition.rs | 78 ++- core/shard/src/lib.rs | 435 ++++++++++++++- core/shard/src/metrics.rs | 48 ++ core/simulator/src/lib.rs | 488 +++++++++++++++- core/simulator/src/replica.rs | 15 +- 13 files changed, 2268 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index a447f0387..b7e276e2b 100644 --- a/README.md +++ b/README.md @@ -284,6 +284,17 @@ When config file is not found, the default values from embedded `config.toml` fi Topic creation accepts two independent policies: `durability` for message acknowledgments and `consumer_offset_durability` for explicit offset stores and deletes. Both default to `replicated`. This means VSR quorum commit without waiting for stable storage. `persisted` also requires recoverable stable-storage copies on the replication quorum. Both policies normally store data on disk. Poll auto-commit remains asynchronous and is not covered by the poll response's completion. +What each policy survives in a cluster of `2f+1` replicas. A partition gets the `persisted` guarantees only when both `durability` and `consumer_offset_durability` are `persisted`. If either is `replicated`, the `replicated` rules below apply to the whole partition. + +- `persisted` keeps every acknowledged write through any pattern of crashes and restarts, including power loss of the whole cluster. Use it when an acknowledged write must survive any single crash. +- `replicated` keeps every acknowledged write as long as a majority of replicas never lose unflushed writes at the same time. +- 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`, more than `f` such crashes at once can lose the most recently acknowledged writes. +- 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 binary polls sent to it fail with `TransientNotAccepted`, which clients retry. HTTP polls are answered from the node's local copy and can return older data. + When no primary answers, the restarted replicas hold an election and mark their logs as possibly incomplete. A new view starts only from a majority of replicas whose logs are not marked, 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, but a replica that never returns halts the partition indefinitely (no operator override yet). + Older builds neither mark their logs nor honor the mark, so during a rolling upgrade these guarantees hold only once every replica runs a build that does. +- When a replica adopts a view whose headers contradict partition history it already applied, it sets that partition's local files aside in `.fenced.N` and fetches it again from a peer instead of stopping the node. A contradiction outside the headers the view carries, and a WAL append whose parent check fails, still stop the node. + The data directory is configured with `path` or `IGGY_PATH`. The layout beneath it is `streams/<stream>/topics/<topic>/partitions/<partition>`, with fixed directory names. The HTTP `Iggy-Durability` header reports `replicated` or `persisted` for awaited writes, and `none` for early dispatch acceptance. diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs index 91b5318f2..a60a485c8 100644 --- a/core/binary_protocol/src/consensus/header.rs +++ b/core/binary_protocol/src/consensus/header.rs @@ -1530,7 +1530,16 @@ pub struct DoViewChangeHeader { pub group: u64, /// View when status was last normal (key for log selection). pub log_view: u32, - pub reserved: [u8; 68], + /// Bit 0, [`Self::FLAG_LOG_UNTRUSTED`]: the sender recovered a `replicated` + /// log after a restart and has not adopted a settled view since, so ops past + /// its head may be ones it acknowledged and lost. Such a log can neither be + /// canonical nor nack anything while a quorum of trusted senders reports. + /// + /// Carved from the head of the former `reserved` region. Zero reads as + /// "trusted", which is what a build predating the flag sends, so the + /// protection holds once every replica runs a build that sets it. + pub flags: u8, + pub reserved: [u8; 67], /// Bit `i` set means the sender proves it never prepared suffix entry `i`, so /// that entry never reached a replication quorum through this replica. A new /// primary may truncate only once `quorum_nack_prepare` senders nack an entry; @@ -1560,14 +1569,19 @@ const _: () = { == offset_of!(DoViewChangeHeader, reserved_frame) + size_of::<[u8; 66]>() ); // op/commit/group/log_view keep their pre-bitset offsets. - assert!(offset_of!(DoViewChangeHeader, reserved) == 156); + assert!(offset_of!(DoViewChangeHeader, op) == 128); + assert!(offset_of!(DoViewChangeHeader, commit) == 136); + assert!(offset_of!(DoViewChangeHeader, group) == 144); + assert!(offset_of!(DoViewChangeHeader, log_view) == 152); + assert!(offset_of!(DoViewChangeHeader, flags) == 156); + assert!(offset_of!(DoViewChangeHeader, reserved) == 157); // Both bitsets are last and 16-aligned, so the struct has no padding // (`NoUninit` would reject any). assert!(offset_of!(DoViewChangeHeader, nack_bitset) % 16 == 0); assert!(offset_of!(DoViewChangeHeader, present_bitset) % 16 == 0); assert!( offset_of!(DoViewChangeHeader, nack_bitset) - == offset_of!(DoViewChangeHeader, reserved) + size_of::<[u8; 68]>() + == offset_of!(DoViewChangeHeader, reserved) + size_of::<[u8; 67]>() ); assert!(offset_of!(DoViewChangeHeader, present_bitset) + size_of::<u128>() == HEADER_SIZE); }; @@ -1623,6 +1637,15 @@ impl ConsensusHeader for DoViewChangeHeader { "commit cannot exceed op".to_string(), )); } + // A bit this build cannot read may carry a claim the merge must honor, and + // reading past it could seat an untrusted log as canonical. Dropping the + // vote costs only availability. + if self.flags & !Self::FLAGS_KNOWN != 0 { + return Err(ConsensusError::InvalidField(format!( + "do_view_change: unknown flag bits {:#04x}", + self.flags & !Self::FLAGS_KNOWN + ))); + } let suffix_len = self.suffix_len()?; // Bits past the suffix describe entries never sent: unchecked, a peer could // smuggle a nack for an op the new primary would then truncate. @@ -1639,6 +1662,17 @@ impl ConsensusHeader for DoViewChangeHeader { } impl DoViewChangeHeader { + /// See [`Self::flags`]. + pub const FLAG_LOG_UNTRUSTED: u8 = 1; + /// Every flag bit this build understands. + pub const FLAGS_KNOWN: u8 = Self::FLAG_LOG_UNTRUSTED; + + /// Whether the sender marked its log untrusted. See [`Self::flags`]. + #[must_use] + pub const fn log_untrusted(&self) -> bool { + self.flags & Self::FLAG_LOG_UNTRUSTED != 0 + } + /// Number of `PrepareHeader`s in the body. /// /// Zero is valid and means "no suffix": a replica with nothing uncommitted @@ -3044,6 +3078,59 @@ mod tests { assert_eq!(prepare.verify_frame(), Ok(())); } + /// A `DoViewChange` as a sender stamps it, with an empty suffix. + fn do_view_change(flags: u8) -> DoViewChangeHeader { + let mut header = control_header::<DoViewChangeHeader>(); + header.view = 7; + header.log_view = 5; + header.op = 42; + header.commit = 40; + header.group = 9; + header.flags = flags; + header.seal(); + header + } + + #[test] + fn given_untrusted_flag_when_round_tripped_should_keep_flag_and_field_offsets() { + let header = do_view_change(DoViewChangeHeader::FLAG_LOG_UNTRUSTED); + let mut frame = AlignedFrame([0u8; HEADER_SIZE]); + frame.0.copy_from_slice(bytemuck::bytes_of(&header)); + + assert_eq!(frame.0[156], DoViewChangeHeader::FLAG_LOG_UNTRUSTED); + assert_eq!(frame.0[128..136], 42u64.to_le_bytes()); + assert_eq!(frame.0[136..144], 40u64.to_le_bytes()); + assert_eq!(frame.0[144..152], 9u64.to_le_bytes()); + assert_eq!(frame.0[152..156], 5u32.to_le_bytes()); + assert!(frame.0[157..224].iter().all(|&byte| byte == 0)); + + let decoded = bytemuck::checked::try_from_bytes::<DoViewChangeHeader>(&frame.0) + .expect("a stamped header is a valid bit pattern"); + assert_eq!(*decoded, header); + assert!(decoded.log_untrusted()); + assert_eq!(decoded.verify_frame(), Ok(())); + assert_eq!(decoded.validate(), Ok(())); + } + + #[test] + fn given_zero_flags_when_decoded_should_read_as_trusted() { + // What a build predating the flag sends in the same bytes. + let header = do_view_change(0); + assert!(!header.log_untrusted()); + assert_eq!(header.validate(), Ok(())); + } + + #[test] + fn given_unknown_flag_bit_when_validating_should_reject() { + for bit in 1..u8::BITS { + let header = do_view_change(DoViewChangeHeader::FLAG_LOG_UNTRUSTED | (1 << bit)); + assert!( + matches!(header.validate(), Err(ConsensusError::InvalidField(_))), + "flag bit {bit} has no meaning yet and must not be read past" + ); + } + } + #[test] fn all_headers_are_256_bytes() { assert_eq!(size_of::<GenericHeader>(), 256); diff --git a/core/consensus/src/dvc_merge.rs b/core/consensus/src/dvc_merge.rs index 0f8dfca4d..81e69747b 100644 --- a/core/consensus/src/dvc_merge.rs +++ b/core/consensus/src/dvc_merge.rs @@ -27,7 +27,9 @@ #[cfg(test)] use crate::view_change_quorum::DvcSuffix; -use crate::view_change_quorum::{DvcQuorumArray, StoredDvc, dvc_count, dvc_iter}; +use crate::view_change_quorum::{ + DvcQuorumArray, StoredDvc, dvc_count, dvc_iter, dvc_quorum_array_empty, dvc_record, +}; use iggy_binary_protocol::{CHECKSUM_UNSEALED, PrepareHeader}; /// Sizes the merge needs from the replica. @@ -47,6 +49,14 @@ pub struct MergeQuorums { /// never made. Every node's depth is pinned below `DVC_HEADERS_MAX`, so the /// ceiling holds for all of them. pub prepare_queue_ceiling: u64, + /// Whether a sender whose head stops below an op counts as nacking it. + /// + /// Sound only when every sender's log holds every op it acknowledged. A + /// replica that acknowledges a prepare before writing it can restart with a + /// shorter log and so "never prepared" an op it helped commit. Off, such ops + /// stay unless an explicit or conflicting-header nack quorum discards them, + /// so the merge keeps more history, never less. + pub absence_nacks: bool, } /// What the collected DVCs say about starting the view. @@ -177,6 +187,7 @@ fn tally_op<'a>( canonical_senders: &[&'a StoredDvc], canonical_log_view: u32, op: u64, + absence_nacks: bool, ) -> OpVerdict<'a> { let CanonicalAt { header: canonical, @@ -188,7 +199,9 @@ fn tally_op<'a>( for dvc in dvc_iter(quorum) { // The sender's log stops below this op, so it never prepared it. if dvc.op < op { - nacks += 1; + if absence_nacks { + nacks += 1; + } continue; } let Some(index) = dvc.suffix.index_of(dvc.op, op) else { @@ -393,11 +406,70 @@ fn committed_elsewhere( Some(extra) } +/// Decide the new view's log from `DoViewChange`s whose senders may have lost +/// ops they acknowledged, as a sender with [`StoredDvc::log_untrusted`] can. +/// +/// Such a sender acknowledged prepares before writing them and then restarted, +/// so its head can stop below ops it helped commit, and its `log_view` says +/// nothing about how much of that view's log survived. Two quorums are safe: +/// +/// * `view_change` trusted senders, merged alone as plain VSR. A committed op +/// was acknowledged by a replication quorum, which any `view_change` replicas +/// intersect, so some trusted sender acknowledged it. Having lost nothing, that +/// sender still holds it or adopted a later view whose log does, and its +/// absence nacks are honest. Untrusted senders contribute no log and no nacks. +/// * Every replica, absence nacks off. No holder of a committed op is left out, +/// and a short head proves nothing when any sender may have lost its tail. +/// +/// Trusted senders alone may leave an op undecided that an untrusted sender's +/// body would settle, so once every replica has reported the second rule gets +/// its turn rather than waiting on DVCs that will never come. +/// +/// `quorums.view_change` is the plain quorum and `quorums.absence_nacks` is +/// ignored: both follow from which rule applies. +#[must_use] +pub fn merge_view_change(quorum: &DvcQuorumArray, quorums: MergeQuorums) -> MergeOutcome { + let received = dvc_count(quorum); + let trusted = dvc_iter(quorum).filter(|dvc| !dvc.log_untrusted).count(); + if trusted >= quorums.view_change { + let trusted_quorums = MergeQuorums { + absence_nacks: true, + ..quorums + }; + let outcome = if trusted == received { + merge_dvc_quorum(quorum, trusted_quorums) + } else { + let mut trusted_only = dvc_quorum_array_empty(); + for dvc in dvc_iter(quorum).filter(|dvc| !dvc.log_untrusted) { + dvc_record(&mut trusted_only, dvc.clone()); + } + merge_dvc_quorum(&trusted_only, trusted_quorums) + }; + if matches!(outcome, MergeOutcome::Ready(_)) || received < quorums.replica_count { + return outcome; + } + } else if received < quorums.replica_count { + return MergeOutcome::AwaitingQuorum; + } + merge_dvc_quorum( + quorum, + MergeQuorums { + view_change: quorums.replica_count, + absence_nacks: false, + ..quorums + }, + ) +} + /// Decide whether the new view can start, and with what log. /// /// Walks every op that might be uncommitted, from the proven commit point to the /// highest a canonical sender reported, stopping at the first proved dead. #[must_use] +#[expect( + clippy::too_many_lines, + reason = "one walk over the op range; splitting it hides the stop conditions" +)] pub fn merge_dvc_quorum(quorum: &DvcQuorumArray, quorums: MergeQuorums) -> MergeOutcome { let received = dvc_count(quorum); if received < quorums.view_change { @@ -440,7 +512,13 @@ pub fn merge_dvc_quorum(quorum: &DvcQuorumArray, quorums: MergeQuorums) -> Merge // commit point is scanned so the adopted log is anchored on a servable header. let mut op = commit_max.max(1); while op <= op_head_max { - let verdict = tally_op(quorum, &canonical_senders, canonical_log_view, op); + let verdict = tally_op( + quorum, + &canonical_senders, + canonical_log_view, + op, + quorums.absence_nacks, + ); if verdict.nacks >= quorums.nack_prepare { if op <= commit_max { @@ -540,7 +618,7 @@ pub fn suffix_all_present(headers: Vec<PrepareHeader>) -> DvcSuffix { mod tests { use super::*; use crate::DVC_HEADERS_MAX; - use crate::view_change_quorum::{dvc_blank, dvc_quorum_array_empty, dvc_record}; + use crate::view_change_quorum::dvc_blank; use iggy_binary_protocol::{Command, Operation}; /// Three replicas: replication 2, view-change 2, nack 2. @@ -550,6 +628,7 @@ mod tests { nack_prepare: 2, replica_count: 3, prepare_queue_ceiling: 32, + absence_nacks: true, } } @@ -585,6 +664,7 @@ mod tests { op, commit, suffix, + log_untrusted: false, } } @@ -1012,4 +1092,208 @@ mod tests { "the head header's commit field is a commit_max lower bound" ); } + + #[test] + fn given_untrusted_absence_when_merging_should_keep_ops_only_one_sender_holds() { + // Replica 0 never restarted and holds op 10. Replicas 1 and 2 restarted + // from logs ending at 9 and 5, having acknowledged op 10 before it reached + // their disks. Their short heads are not proof they never prepared it. + let mut quorum = dvc_quorum_array_empty(); + dvc_record( + &mut quorum, + dvc(0, 1, 10, 8, suffix_all_present(suffix_headers(8, 10, 1))), + ); + dvc_record( + &mut quorum, + dvc(1, 1, 9, 7, suffix_all_present(suffix_headers(7, 9, 1))), + ); + dvc_record( + &mut quorum, + dvc(2, 1, 5, 4, suffix_all_present(suffix_headers(4, 5, 1))), + ); + + let MergeOutcome::Ready(trusting) = merge_dvc_quorum(&quorum, quorums_r3()) else { + panic!("a quorum with absence nacks decides"); + }; + assert_eq!(trusting.op_head, 9, "two absent heads nack op 10"); + + let quorums = MergeQuorums { + absence_nacks: false, + ..quorums_r3() + }; + let MergeOutcome::Ready(untrusting) = merge_dvc_quorum(&quorum, quorums) else { + panic!("op 10 has a servable copy, so the view can start"); + }; + assert_eq!( + untrusting.op_head, 10, + "an op some sender can serve survives senders that merely lack it" + ); + } + + fn untrusted(dvc: StoredDvc) -> StoredDvc { + StoredDvc { + log_untrusted: true, + ..dvc + } + } + + /// What a recovering replica barred from electing sends. + fn empty_vote(replica: u8) -> StoredDvc { + untrusted(dvc(replica, 0, 0, 0, DvcSuffix::empty())) + } + + #[test] + fn given_quorum_of_trusted_senders_when_an_untrusted_one_has_a_higher_log_view_should_merge_the_trusted_log() + { + // Replica 2 led view 2 and restarted having lost its tail. Its log at + // log_view 2 ends at 6 with a header no trusted sender holds, and would be + // canonical by log_view. Replicas 0 and 1 trust their view-1 logs to op 9. + let mut quorum = dvc_quorum_array_empty(); + dvc_record( + &mut quorum, + dvc(0, 1, 9, 4, suffix_all_present(suffix_headers(4, 9, 1))), + ); + dvc_record( + &mut quorum, + untrusted(dvc(2, 2, 6, 4, suffix_all_present(suffix_headers(4, 6, 2)))), + ); + assert_eq!( + merge_view_change(&quorum, quorums_r3()), + MergeOutcome::AwaitingQuorum, + "one trusted sender is short of the view-change quorum" + ); + + dvc_record( + &mut quorum, + dvc(1, 1, 8, 4, suffix_all_present(suffix_headers(4, 8, 1))), + ); + let MergeOutcome::Ready(merged) = merge_view_change(&quorum, quorums_r3()) else { + panic!("two trusted senders are a quorum"); + }; + assert_eq!(merged.op_head, 9); + assert!( + merged.headers.iter().all(|header| header.view == 1), + "the untrusted sender's higher log_view made its headers canonical" + ); + } + + #[test] + fn given_quorum_of_trusted_senders_when_merging_should_take_absence_nacks_only_from_them() { + // Replica 2 is untrusted and stops at 5: its short head must not nack + // op 9 alongside replica 1's honest one. + let mut quorum = dvc_quorum_array_empty(); + dvc_record( + &mut quorum, + dvc(0, 1, 9, 4, suffix_all_present(suffix_headers(4, 9, 1))), + ); + dvc_record( + &mut quorum, + dvc(1, 1, 8, 4, suffix_all_present(suffix_headers(4, 8, 1))), + ); + dvc_record( + &mut quorum, + untrusted(dvc(2, 1, 5, 4, suffix_all_present(suffix_headers(4, 5, 1)))), + ); + + let MergeOutcome::Ready(merged) = merge_view_change(&quorum, quorums_r3()) else { + panic!("two trusted senders are a quorum"); + }; + assert_eq!(merged.op_head, 9); + } + + #[test] + fn given_fewer_trusted_senders_than_a_quorum_should_wait_for_every_replica() { + // The trusted lagging replica 1 and the untrusted replica 2 are a quorum by + // count, and replica 0, the only holder of ops 6..=9, has not reported. + let mut quorum = dvc_quorum_array_empty(); + dvc_record( + &mut quorum, + dvc(1, 1, 5, 4, suffix_all_present(suffix_headers(4, 5, 1))), + ); + dvc_record( + &mut quorum, + untrusted(dvc(2, 2, 7, 4, suffix_all_present(suffix_headers(4, 7, 2)))), + ); + assert_eq!( + merge_view_change(&quorum, quorums_r3()), + MergeOutcome::AwaitingQuorum + ); + + dvc_record( + &mut quorum, + dvc(0, 2, 9, 4, suffix_all_present(suffix_headers(4, 9, 2))), + ); + let MergeOutcome::Ready(merged) = merge_view_change(&quorum, quorums_r3()) else { + panic!("every replica reported"); + }; + assert_eq!( + merged.op_head, 9, + "the view dropped ops only replica 0 holds" + ); + } + + #[test] + fn given_every_replica_with_mixed_flags_when_trusted_merge_is_undecided_should_merge_all() { + // Both trusted senders reach op 9 but sent no suffix, so they abstain on + // every uncommitted op. Only the untrusted sender holds the headers and + // bodies, and once nobody is left to report the all-replica rule decides. + let mut quorum = dvc_quorum_array_empty(); + dvc_record(&mut quorum, dvc(0, 1, 9, 4, DvcSuffix::empty())); + dvc_record(&mut quorum, dvc(1, 1, 9, 4, DvcSuffix::empty())); + assert_eq!( + merge_view_change(&quorum, quorums_r3()), + MergeOutcome::AwaitingRepair { undecided_op: 4 } + ); + + dvc_record( + &mut quorum, + untrusted(dvc(2, 1, 9, 4, suffix_all_present(suffix_headers(4, 9, 1)))), + ); + let MergeOutcome::Ready(merged) = merge_view_change(&quorum, quorums_r3()) else { + panic!("the untrusted sender serves op 9"); + }; + assert_eq!(merged.op_head, 9); + } + + #[test] + fn given_an_empty_vote_when_every_replica_reports_should_count_it_and_keep_every_op() { + let mut quorum = dvc_quorum_array_empty(); + dvc_record( + &mut quorum, + untrusted(dvc(0, 1, 9, 4, suffix_all_present(suffix_headers(4, 9, 1)))), + ); + dvc_record( + &mut quorum, + untrusted(dvc(1, 1, 5, 4, suffix_all_present(suffix_headers(4, 5, 1)))), + ); + assert_eq!( + merge_view_change(&quorum, quorums_r3()), + MergeOutcome::AwaitingQuorum + ); + + dvc_record(&mut quorum, empty_vote(2)); + let MergeOutcome::Ready(merged) = merge_view_change(&quorum, quorums_r3()) else { + panic!("every replica reported"); + }; + assert_eq!(merged.op_head, 9, "the empty vote nacked an op"); + assert_eq!(merged.commit_max, 4); + } + + #[test] + fn given_only_empty_votes_and_untrusted_logs_should_never_adopt_the_empty_log() { + // Every sender at log_view 0 makes the empty vote a canonical sender by + // log_view; holding nothing, it must still contribute nothing. + let mut quorum = dvc_quorum_array_empty(); + dvc_record( + &mut quorum, + untrusted(dvc(0, 0, 3, 0, suffix_all_present(suffix_headers(1, 3, 0)))), + ); + dvc_record(&mut quorum, empty_vote(1)); + dvc_record(&mut quorum, empty_vote(2)); + + let MergeOutcome::Ready(merged) = merge_view_change(&quorum, quorums_r3()) else { + panic!("every replica reported"); + }; + assert_eq!(merged.op_head, 3); + } } diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index f85052c1c..b1a41bd9c 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -22,7 +22,7 @@ use crate::{ MergeOutcome, MergeQuorums, MergedLog, Pipeline, PlaneKind, PrepareLogEvent, Project, ReplicaLogContext, SimEventKind, StoredDvc, ViewChangeLogEvent, ViewChangeReason, VsrState, dvc_count, dvc_iter, dvc_quorum_array_empty, dvc_record, dvc_reset, dvc_suffix_decode, - emit_replica_event, emit_sim_event, merge_dvc_quorum, seal_prepare_checksum, + emit_replica_event, emit_sim_event, merge_view_change, seal_prepare_checksum, }; use bit_set::BitSet; use clock::{Clock, IggySystemClock}; @@ -1075,6 +1075,8 @@ pub enum VsrAction { /// this replica's own `StoredDvc`: a merge seeing two versions of one /// sender's suffix could adopt a header no replica holds. suffix: DvcSuffix, + /// Stamped as `DoViewChangeHeader::FLAG_LOG_UNTRUSTED`. + log_untrusted: bool, }, /// Broadcast a `RequestStartView` probe (recovering replica asking for /// the current view's `StartView`; only that view's primary answers). @@ -1148,6 +1150,26 @@ pub enum VsrAction { }, } +impl VsrAction { + /// Whether this is the log-less `DoViewChange` a recovering replica barred + /// from electing sends. It vouches for no history, so a partition awaiting + /// a state transfer may still send it. + #[must_use] + pub fn is_empty_vote(&self) -> bool { + matches!( + self, + Self::SendDoViewChange { + log_view: 0, + op: 0, + commit: 0, + log_untrusted: true, + suffix, + .. + } if suffix.is_empty() + ) + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum PrepareOkOutcome { Accepted { @@ -1274,6 +1296,24 @@ where /// built-in default. probe_attempts_max: Cell<u32>, recovery_election_allowed: Cell<bool>, + /// False when this replica may acknowledge a prepare before the write that + /// recovery reads back reaches stable storage, so a restart can come back + /// with a log shorter than what it acknowledged. See + /// [`Self::set_recovered_log_complete`]. + recovered_log_complete: Cell<bool>, + /// This replica left `Recovering` through the probe-exhaustion election + /// while [`Self::recovered_log_complete`] was false: its log may lack ops it + /// acknowledged, and no settled primary has vouched for it since. + /// + /// Every `DoViewChange` it sends says so, and no merge counts such a log + /// toward a quorum: a view starts from a quorum of trusted senders, or from + /// every replica (see `merge_view_change`). Cleared by adopting a + /// `StartView` or by starting a view as primary, whose merged log either + /// rule makes safe. + untrusted_log: Cell<bool>, + /// View-change escalations since [`Self::untrusted_log`] was last set, for + /// rate-limiting the halt warning. + untrusted_view_change_timeouts: Cell<u32>, /// This replica's own uncommitted suffix, with the head, commit point and /// mutation count the journal was at when it was read. @@ -1665,6 +1705,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { probe_attempts: Cell::new(0), probe_attempts_max: Cell::new(PROBE_ATTEMPTS_MAX), recovery_election_allowed: Cell::new(true), + recovered_log_complete: Cell::new(true), + untrusted_log: Cell::new(false), + untrusted_view_change_timeouts: Cell::new(0), local_dvc_suffix: RefCell::new(None), journal_mutations: Cell::new(0), pending_view_log: RefCell::new(None), @@ -1750,6 +1793,32 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.recovery_election_allowed.get() } + /// Declare whether a restart recovers every prepare this replica + /// acknowledged. `true` (the default) when every acknowledgement waits for + /// stable storage; `false` when some are sent once the write is queued, as + /// `replicated` partition durability does. + /// + /// With `false`, a replica whose view probe goes unanswered still elects, + /// since a full-cluster restart has nobody to answer it, but its log is + /// untrusted until a settled primary vouches for it: its `DoViewChange` + /// says so, and no view starts on it short of a trusted quorum or every + /// replica reporting. See [`Self::has_untrusted_log`]. + pub fn set_recovered_log_complete(&self, complete: bool) { + self.recovered_log_complete.set(complete); + } + + #[must_use] + pub const fn recovered_log_complete(&self) -> bool { + self.recovered_log_complete.get() + } + + /// Whether this replica elected on a recovered log that may lack ops it + /// acknowledged and has not yet been synced by a settled primary. + #[must_use] + pub const fn has_untrusted_log(&self) -> bool { + self.untrusted_log.get() + } + pub fn init(&self) { self.status.set(Status::Normal); let mut timeouts = self.timeouts.borrow_mut(); @@ -2709,6 +2778,22 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.set_state_transfer_stage(StateTransferStage::Idle); } self.finish_view_probe(); + // "Nobody answered" does not prove nobody is alive: a + // survivor mid view change answers no probe. If this log + // can lack acknowledged ops, a quorum of restarted + // replicas could otherwise elect past a survivor holding + // committed ops that only it still has. + if !self.recovered_log_complete.get() { + tracing::info!( + replica = self.replica, + namespace_raw = self.group, + view = self.view.get(), + "view probe exhausted on a log that may lack acknowledged ops; \ + electing, but marking the log untrusted in every DoViewChange" + ); + self.untrusted_log.set(true); + self.untrusted_view_change_timeouts.set(0); + } actions.extend( self.start_election(plane, ViewChangeReason::ViewProbeUnanswered), ); @@ -2936,6 +3021,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { if self.status.get() != Status::ViewChange { return Vec::new(); } + if self.untrusted_log.get() { + self.warn_untrusted_halt(); + } self.enter_view_change( plane, @@ -2944,6 +3032,47 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { ) } + /// No view starts without a quorum of trusted logs or every replica + /// reporting, which looks like an ordinary stuck view change unless it says + /// so. Logged on the 1st, 2nd, 4th ... 64th escalation and every 64th after, + /// so a halt that lasts hours stays visible without a line per timeout. + fn warn_untrusted_halt(&self) { + const STEADY_INTERVAL: u32 = 64; + let timeouts = self.untrusted_view_change_timeouts.get().saturating_add(1); + self.untrusted_view_change_timeouts.set(timeouts); + let due = if timeouts <= STEADY_INTERVAL { + timeouts.is_power_of_two() + } else { + timeouts.is_multiple_of(STEADY_INTERVAL) + }; + if !due { + return; + } + let view = self.view.get(); + let primary = self.primary_index(view); + let awaited_do_view_change: Vec<u8> = if self.is_primary_for_view(view) { + let received = self.do_view_change_from_all_replicas.borrow(); + (0..self.replica_count) + .filter(|replica| { + *replica != self.replica && received[usize::from(*replica)].is_none() + }) + .collect() + } else { + Vec::new() + }; + tracing::warn!( + replica = self.replica, + namespace_raw = self.group, + view, + primary, + view_change_timeouts = timeouts, + ?awaited_do_view_change, + "view change halted: this replica elected on a log that may lack acknowledged ops, \ + so no view starts until a quorum of replicas with intact logs, or every replica, \ + reports" + ); + } + /// Collect uncommitted pipeline entries that should be retransmitted. /// /// Returns `(PrepareHeader, Vec<u8>)` pairs: each op that hasn't reached @@ -3087,9 +3216,13 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // A recovering replica is quorum-invisible: it lost (or cannot trust) // its durable state, so it must not vote history into existence. The // election proceeds among the peers; its conclusion reaches this - // replica via StartView, which recovery accepts. + // replica via StartView, which recovery accepts. The empty vote + // carries no history, so it is no exception. if self.status.get() == Status::Recovering { - return Vec::new(); + return self + .empty_do_view_change(plane, header.view) + .into_iter() + .collect(); } let from_replica = header.replica; let msg_view = header.view; @@ -3149,6 +3282,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { op: current_op, commit, suffix: self.local_dvc_suffix(), + log_untrusted: self.untrusted_log.get(), }; dvc_record( &mut self.do_view_change_from_all_replicas.borrow_mut(), @@ -3197,7 +3331,50 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { commit: self.dvc_commit(), group: self.group, suffix: self.local_dvc_suffix(), + log_untrusted: self.untrusted_log.get(), + } + } + + /// The `DoViewChange` a recovering replica barred from electing sends for + /// `view`, or `None` when it must stay silent. + /// + /// Without it a group whose every log is untrusted could never start a + /// view, since that takes every replica reporting and a barred replica + /// waits for a `StartView` instead. The vote holds no log: flagged + /// untrusted, at `log_view` 0 with nothing in it, it nacks nothing, offers + /// nothing and commits nothing, so it counts toward every replica reporting + /// and decides nothing else. It also leaves this replica `Recovering` in its + /// own view and never makes it a candidate: it waits for the `StartView` + /// and the state transfer it was waiting for already. + /// + /// Only where a recovered log can be untrusted. Elsewhere a quorum of + /// trusted peers never needs it. + fn empty_do_view_change(&self, plane: PlaneKind, view: u32) -> Option<VsrAction> { + if self.recovered_log_complete.get() + || self.recovery_election_allowed.get() + || view < self.view.get() + || self.is_primary_for_view(view) + { + return None; } + let action = VsrAction::SendDoViewChange { + view, + target: self.primary_index(view), + log_view: 0, + op: 0, + commit: 0, + group: self.group, + suffix: DvcSuffix::empty(), + log_untrusted: true, + }; + emit_sim_event( + SimEventKind::ControlMessageScheduled, + &ControlActionLogEvent::from_vsr_action( + ReplicaLogContext::from_consensus(self, plane), + &action, + ), + ); + Some(action) } /// Decode a peer's suffix, or `None` to drop the whole `DoViewChange`. @@ -3245,7 +3422,10 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // Quorum-invisible while recovering (see handle_start_view_change): // a recovering replica must not collect DVCs and crown itself. if self.status.get() == Status::Recovering { - return Vec::new(); + return self + .empty_do_view_change(plane, header.view) + .into_iter() + .collect(); } let from_replica = header.replica; let msg_view = header.view; @@ -3296,6 +3476,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { op: current_op, commit, suffix: self.local_dvc_suffix(), + log_untrusted: self.untrusted_log.get(), }; dvc_record( &mut self.do_view_change_from_all_replicas.borrow_mut(), @@ -3310,6 +3491,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { op: msg_op, commit: msg_commit, suffix: msg_suffix, + log_untrusted: header.log_untrusted(), }; dvc_record(&mut self.do_view_change_from_all_replicas.borrow_mut(), dvc); @@ -3670,6 +3852,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.log_view.set(msg_view); self.status.set(Status::Normal); self.ceded_primaryship.set(false); + // The view's log now comes from a settled primary, and repair fills this + // replica's holes from it. + self.untrusted_log.set(false); self.advance_commit_max(msg_commit); self.reset_view_change_state(); @@ -3824,7 +4009,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { fn complete_view_change_as_primary(&self, plane: PlaneKind) -> Vec<VsrAction> { let merged = { let dvc_array = self.do_view_change_from_all_replicas.borrow(); - merge_dvc_quorum(&dvc_array, self.merge_quorums()) + merge_view_change(&dvc_array, self.merge_quorums()) }; let merged = match merged { @@ -3931,7 +4116,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { Vec::new() } - /// Sizes handed to the DVC merge. + /// Sizes handed to the DVC merge. `merge_view_change` raises the quorum to + /// every replica, and drops absence nacks, when too few senders trust their + /// logs. const fn merge_quorums(&self) -> MergeQuorums { MergeQuorums { view_change: self.quorum_view_change(), @@ -3940,6 +4127,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // The cluster-wide ceiling, not `self.prepare_queue_max`: this node's // config says nothing about how deep a peer's pipeline is. prepare_queue_ceiling: PREPARE_QUEUE_CEILING as u64, + absence_nacks: true, } } @@ -4048,6 +4236,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.status.set(Status::Normal); self.ceded_primaryship.set(false); + // Merged from a trusted quorum or from every replica, so no surviving log + // was left out. + self.untrusted_log.set(false); self.sequencer.set_sequence(new_op); // Only place a promotion learns what the merge did to a gated suffix. self.redecide_recovery_barrier(new_op); @@ -6270,6 +6461,7 @@ mod view_source_tests { op, commit, suffix: DvcSuffix::new(headers, 0, present), + log_untrusted: false, } } @@ -6571,7 +6763,7 @@ mod recovery_election_tests { use crate::dvc_merge::suffix_all_present; use crate::test_bus::NoopBus; use crate::view_change_quorum::encode_prepare_headers; - use iggy_binary_protocol::{HEADER_SIZE, Operation}; + use iggy_binary_protocol::{CommitHeader, HEADER_SIZE, Operation}; const REPLICA_COUNT: u8 = 3; /// Primary of `OLD_VIEW`, crashed holding only part of what it committed. @@ -6621,8 +6813,13 @@ mod recovery_election_tests { /// A replica rebooted from a WAL ending at `head`, joining as the boot path /// does when the WAL recovered a frontier: probing, with the fallback - /// election allowed. - fn restarted(replica: u8, head: u64) -> VsrConsensus<NoopBus, LocalPipeline> { + /// election allowed. `log_complete` is false under `replicated` durability, + /// which acknowledges a prepare before its write reaches the disk. + fn restarted( + replica: u8, + head: u64, + log_complete: bool, + ) -> VsrConsensus<NoopBus, LocalPipeline> { let commit = prepare(head).commit; let mut consensus = VsrConsensus::new( 1, @@ -6644,6 +6841,7 @@ mod recovery_election_tests { (commit..=head).rev().map(prepare).collect(), )); consensus.set_recovery_election_allowed(true); + consensus.set_recovered_log_complete(log_complete); consensus.join(JoinMode::ProbeAsBackup { await_state_transfer: false, }); @@ -6680,11 +6878,17 @@ mod recovery_election_tests { Message::<H>::new(HEADER_SIZE + body).transmute_header(|_, header: &mut H| fill(header)) } + struct DeliveredDoViewChange { + from: u8, + view: u32, + log_untrusted: bool, + } + struct Cluster { replicas: Vec<VsrConsensus<NoopBus, LocalPipeline>>, plane: PlaneKind, - /// `(sender, receiver, view)` of every `DoViewChange` delivered. - do_view_changes: Vec<(u8, u8, u32)>, + /// Every `DoViewChange` delivered. + do_view_changes: Vec<DeliveredDoViewChange>, } impl Cluster { @@ -6697,14 +6901,11 @@ mod recovery_election_tests { fn deliver( &mut self, mut outbox: VecDeque<(u8, VsrAction)>, - reachable: impl Fn(u8) -> bool, + reachable: impl Fn(u8, u8) -> bool, ) { while let Some((from, action)) = outbox.pop_front() { - if !reachable(from) { - continue; - } for to in 0..REPLICA_COUNT { - if to == from { + if to == from || !reachable(from, to) { continue; } for reply in self.receive(from, to, &action) { @@ -6722,6 +6923,10 @@ mod recovery_election_tests { } } + #[expect( + clippy::too_many_lines, + reason = "one arm per message kind the harness delivers" + )] fn receive(&mut self, from: u8, to: u8, action: &VsrAction) -> Vec<VsrAction> { let plane = self.plane; let receiver = self.replica(to); @@ -6745,6 +6950,7 @@ mod recovery_election_tests { commit, group, suffix, + log_untrusted, } if *target == to => { let mut dvc = message(suffix.encoded_len(), |header: &mut DoViewChangeHeader| { @@ -6758,11 +6964,18 @@ mod recovery_election_tests { header.op = *op; header.commit = *commit; header.group = *group; + if *log_untrusted { + header.flags = DoViewChangeHeader::FLAG_LOG_UNTRUSTED; + } header.nack_bitset = suffix.nack_bitset(); header.present_bitset = suffix.present_bitset(); }); suffix.encode_into(&mut dvc.as_mut_slice()[HEADER_SIZE..]); - self.do_view_changes.push((from, to, *view)); + self.do_view_changes.push(DeliveredDoViewChange { + from, + view: *view, + log_untrusted: *log_untrusted, + }); self.replica(to).handle_do_view_change( plane, dvc.header(), @@ -6808,6 +7021,27 @@ mod recovery_election_tests { &start_view.as_slice()[HEADER_SIZE..], ) } + // Heartbeats keep a settled view settled; nothing else in this + // harness carries them. + VsrAction::SendCommit { + view, + commit, + group, + timestamp_monotonic, + } => { + let heartbeat = message(0, |header: &mut CommitHeader| { + header.command = Command::Commit; + header.cluster = 1; + header.size = u32::try_from(HEADER_SIZE).unwrap(); + header.replica = from; + header.view = *view; + header.commit = *commit; + header.group = *group; + header.timestamp_monotonic = *timestamp_monotonic; + }); + let _ = receiver.handle_commit(heartbeat.header()); + Vec::new() + } _ => Vec::new(), } } @@ -6823,8 +7057,8 @@ mod recovery_election_tests { let voters: Vec<u8> = self .do_view_changes .iter() - .filter(|(_, _, view)| *view == replica.log_view()) - .map(|(from, _, _)| *from) + .filter(|delivered| delivered.view == replica.log_view()) + .map(|delivered| delivered.from) .chain(std::iter::once(replica.primary_index(replica.log_view()))) .collect::<std::collections::BTreeSet<_>>() .into_iter() @@ -6845,6 +7079,41 @@ mod recovery_election_tests { } } + impl Cluster { + /// Tick every replica `ticks` times, delivering what `reachable` allows + /// and checking the committed-ops invariant after every tick. + fn run(&mut self, ticks: u64, reachable: impl Fn(u8, u8) -> bool) { + for tick in 0..ticks { + let mut outbox = VecDeque::new(); + for index in 0..REPLICA_COUNT { + for action in self.replica(index).tick(self.plane) { + outbox.push_back((index, action)); + } + } + self.deliver(outbox, &reachable); + self.assert_committed_ops_survive(tick); + } + } + + /// Every replica settled in one view newer than `OLD_VIEW`, holding at + /// least `head`, and trusting its log again. + fn assert_settled(&self, head: u64) { + let view = self.replica(0).view(); + for replica in &self.replicas { + assert_eq!( + replica.status(), + Status::Normal, + "replica {} never settled", + replica.replica() + ); + assert_eq!(replica.view(), view, "replicas settled in different views"); + assert!(view > OLD_VIEW); + assert!(replica.sequencer().current_sequence() >= head); + assert!(!replica.has_untrusted_log()); + } + } + } + /// Two replicas restart after a double crash, the primary having lost the /// tail of what it committed. The survivor holding that tail is mid view /// change, so it answers no probe. Its view-change traffic reaches the @@ -6852,13 +7121,12 @@ mod recovery_election_tests { /// their fallback election, which an asynchronous network permits. Whatever /// the restarted pair decide without it must still contain the committed ops. #[test] - #[ignore = "#4476: two restarted replicas elect without the survivor and drop committed ops"] fn given_survivor_in_view_change_when_restarted_pair_elects_should_keep_committed_ops() { let plane = PlaneKind::Partitions; let mut cluster = Cluster { replicas: vec![ - restarted(RESTARTED_PRIMARY, RECOVERED_HEAD), - restarted(LAGGING_BACKUP, LAGGING_HEAD), + restarted(RESTARTED_PRIMARY, RECOVERED_HEAD, false), + restarted(LAGGING_BACKUP, LAGGING_HEAD, false), survivor_in_view_change(plane), ], plane, @@ -6880,8 +7148,494 @@ mod recovery_election_tests { let probing = [RESTARTED_PRIMARY, LAGGING_BACKUP] .iter() .all(|index| cluster.replica(*index).status() == Status::Recovering); - cluster.deliver(outbox, |from| probing || from != SURVIVOR); + cluster.deliver(outbox, |from, _| probing || from != SURVIVOR); cluster.assert_committed_ops_survive(tick); } } + + /// The double crash again, with the survivor's traffic never lost. The + /// restarted pair may not settle a view without it, and once it reports + /// the cluster settles holding every committed op. + #[test] + fn given_survivor_in_view_change_when_it_stays_reachable_should_settle_with_committed_ops() { + let plane = PlaneKind::Partitions; + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, RECOVERED_HEAD, false), + restarted(LAGGING_BACKUP, LAGGING_HEAD, false), + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + } + + /// The survivor is cut off for good. The restarted pair, a quorum by + /// count, must stay unavailable rather than settle on their short logs, and + /// recover as soon as the survivor is back. + #[test] + fn given_survivor_partitioned_when_restarted_pair_elects_should_halt_until_it_returns() { + let plane = PlaneKind::Partitions; + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, RECOVERED_HEAD, false), + restarted(LAGGING_BACKUP, LAGGING_HEAD, false), + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(TICKS, |from, to| from != SURVIVOR && to != SURVIVOR); + + for index in [RESTARTED_PRIMARY, LAGGING_BACKUP] { + let replica = cluster.replica(index); + assert_eq!(replica.status(), Status::ViewChange); + assert!( + replica.has_untrusted_log(), + "the fallback election left replica {index} trusting a log that may lack \ + acknowledged ops" + ); + } + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + } + + /// Every replica restarted together and nobody is left to answer a probe. + /// The fallback election must still settle, on the longest log: under + /// `replicated` durability an op the shorter logs lack may have been + /// acknowledged by them before their crash. + #[test] + fn given_full_cluster_restart_when_every_replica_reports_should_settle_on_the_longest_log() { + let plane = PlaneKind::Partitions; + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + restarted(LAGGING_BACKUP, LAGGING_HEAD, false), + restarted(SURVIVOR, RECOVERED_HEAD, false), + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + } + + /// A full restart where one replica's partition boots without the fallback + /// election, awaiting a state transfer. Every log is untrusted, so a view + /// needs every replica, and the barred replica supplies its share as an + /// empty vote: the view starts on the logs that elected, and the barred + /// replica joins it through the `StartView`. + #[test] + fn given_full_restart_with_one_replica_barred_from_election_when_logs_are_untrusted_should_settle_on_its_empty_vote_without_dropping_ops() + { + let plane = PlaneKind::Partitions; + let barred = restarted(SURVIVOR, COMMITTED, false); + barred.set_recovery_election_allowed(false); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + restarted(LAGGING_BACKUP, LAGGING_HEAD, false), + barred, + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + let empty_votes = cluster + .do_view_changes + .iter() + .filter(|delivered| delivered.from == SURVIVOR) + .count(); + assert!(empty_votes > 0, "the barred replica never voted"); + assert!( + cluster + .do_view_changes + .iter() + .filter(|delivered| delivered.from == SURVIVOR) + .all(|delivered| delivered.log_untrusted), + "an empty vote must never read as a trusted log" + ); + } + + /// The barred replica is the candidate of the first view the others elect. + /// It may vote but never lead: it stays `Recovering` until a `StartView` + /// from another primary, so the others move past its view and settle. + #[test] + fn given_barred_replica_is_the_next_primary_candidate_when_others_elect_should_never_lead() { + let plane = PlaneKind::Partitions; + let barred = restarted(LAGGING_BACKUP, COMMITTED, false); + barred.set_recovery_election_allowed(false); + assert_eq!(barred.primary_index(OLD_VIEW + 1), LAGGING_BACKUP); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + barred, + restarted(SURVIVOR, RECOVERED_HEAD, false), + ], + plane, + do_view_changes: Vec::new(), + }; + + for tick in 0..TICKS { + let mut outbox = VecDeque::new(); + for index in 0..REPLICA_COUNT { + for action in cluster.replica(index).tick(plane) { + outbox.push_back((index, action)); + } + } + cluster.deliver(outbox, |_, _| true); + cluster.assert_committed_ops_survive(tick); + let barred = cluster.replica(LAGGING_BACKUP); + assert!( + barred.status() == Status::Recovering + || (barred.status() == Status::Normal && !barred.is_primary()), + "tick {tick}: the barred replica left Recovering on its own: status {:?}, view {}", + barred.status(), + barred.view() + ); + } + + cluster.assert_settled(COMMITTED); + assert!(!cluster.replica(LAGGING_BACKUP).is_primary()); + } + + /// Cut off from everyone but the barred replica's own SVC and DVC traffic, + /// an empty voter stays `Recovering` and in its own view: voting is not + /// joining. + #[test] + fn given_barred_replica_when_it_only_hears_a_view_change_should_stay_recovering() { + let plane = PlaneKind::Partitions; + let barred = restarted(SURVIVOR, COMMITTED, false); + barred.set_recovery_election_allowed(false); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + restarted(LAGGING_BACKUP, COMMITTED, false), + barred, + ], + plane, + do_view_changes: Vec::new(), + }; + let view = cluster.replica(SURVIVOR).view(); + + // The barred replica hears everything, its votes reach nobody. + cluster.run(TICKS, |from, _| from != SURVIVOR); + + let barred = cluster.replica(SURVIVOR); + assert_eq!(barred.status(), Status::Recovering); + assert_eq!(barred.view(), view); + assert!( + cluster + .do_view_changes + .iter() + .all(|delivered| delivered.from != SURVIVOR) + ); + for index in [RESTARTED_PRIMARY, LAGGING_BACKUP] { + assert_eq!(cluster.replica(index).status(), Status::ViewChange); + assert!(cluster.replica(index).has_untrusted_log()); + } + } + + /// A restarted replica still probing may yet elect on its log, which can + /// hold ops nobody else has. It must not cast an empty vote that would let + /// the others start a view without that log. + #[test] + fn given_full_restart_when_a_replica_still_probes_should_wait_for_its_log() { + let plane = PlaneKind::Partitions; + let slow = restarted(SURVIVOR, COMMITTED, false); + slow.set_probe_attempts_max(4 * PROBE_ATTEMPTS_MAX); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, RECOVERED_HEAD, false), + restarted(LAGGING_BACKUP, LAGGING_HEAD, false), + slow, + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(2 * TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + } + + /// Persisted recovery never needs the barred replica, so it stays silent + /// and the trusted pair settle as a plain quorum. + #[test] + fn given_complete_logs_when_a_replica_is_barred_should_send_no_empty_vote() { + let plane = PlaneKind::Partitions; + let barred = restarted(SURVIVOR, COMMITTED, true); + barred.set_recovery_election_allowed(false); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, true), + restarted(LAGGING_BACKUP, LAGGING_HEAD, true), + barred, + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + assert!( + cluster + .do_view_changes + .iter() + .all(|delivered| delivered.from != SURVIVOR && !delivered.log_untrusted) + ); + } + + /// `persisted` acknowledges only what reached the disk, so a restarted log + /// holds every op it acknowledged and the restarted pair keep the fast + /// fallback: they settle without waiting for the survivor. + #[test] + fn given_complete_recovered_logs_when_survivor_is_silent_should_elect_without_it() { + let plane = PlaneKind::Partitions; + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, true), + restarted(LAGGING_BACKUP, LAGGING_HEAD, true), + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(TICKS, |from, to| from != SURVIVOR && to != SURVIVOR); + + let settled: Vec<u8> = [RESTARTED_PRIMARY, LAGGING_BACKUP] + .into_iter() + .filter(|index| { + let replica = cluster.replica(*index); + replica.status() == Status::Normal && replica.log_view() > OLD_VIEW + }) + .collect(); + assert_eq!(settled, vec![RESTARTED_PRIMARY, LAGGING_BACKUP]); + assert!( + cluster + .replicas + .iter() + .all(|replica| !replica.has_untrusted_log()) + ); + } + + /// An untrusted primary candidate holding a quorum of `DoViewChange`s waits + /// for the last one, and the merge keeps the op only that sender holds even + /// though both restarted senders lack it. + #[test] + fn given_untrusted_primary_candidate_when_a_quorum_reports_should_wait_for_every_replica() { + let plane = PlaneKind::Partitions; + let candidate = restarted(LAGGING_BACKUP, LAGGING_HEAD, false); + while candidate.status() == Status::Recovering { + candidate.tick(plane); + } + let view = candidate.view(); + assert_eq!(candidate.primary_index(view), LAGGING_BACKUP); + assert!(candidate.has_untrusted_log()); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, RECOVERED_HEAD, false), + candidate, + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + }; + let do_view_change = + |head: u64, commit: u64, log_untrusted: bool| VsrAction::SendDoViewChange { + view, + target: LAGGING_BACKUP, + log_view: OLD_VIEW, + op: head, + commit, + group: group(), + suffix: suffix_all_present((commit..=head).rev().map(prepare).collect()), + log_untrusted, + }; + + cluster.receive( + RESTARTED_PRIMARY, + LAGGING_BACKUP, + &do_view_change(RECOVERED_HEAD, RECOVERED_HEAD - 1, true), + ); + assert!( + !cluster.replica(LAGGING_BACKUP).view_log_is_pending(), + "decided a view from {} of {REPLICA_COUNT} DoViewChanges", + cluster.replica(LAGGING_BACKUP).quorum_view_change() + ); + + cluster.receive( + SURVIVOR, + LAGGING_BACKUP, + &do_view_change(COMMITTED, COMMITTED - 1, false), + ); + let candidate = cluster.replica(LAGGING_BACKUP); + assert_eq!( + candidate.with_pending_view_log(|log| log.op_head), + Some(COMMITTED), + "the merge dropped ops only the survivor holds" + ); + candidate.start_pending_view(plane); + assert_eq!(candidate.status(), Status::Normal); + assert!(!candidate.has_untrusted_log()); + } + + /// The lagging backup never restarted, so it trusts its log and is the + /// candidate of the next view. The restarted primary's `DoViewChange` + /// completes a quorum by count while the survivor holding the committed tail + /// is cut off. Flagged untrusted, it does not complete a trusted one, so the + /// candidate waits for the survivor rather than dropping the tail. + #[test] + fn given_trusted_lagging_primary_when_restarted_replica_completes_its_quorum_should_keep_committed_ops() + { + let plane = PlaneKind::Partitions; + let lagging = { + let mut consensus = VsrConsensus::new( + 1, + LAGGING_BACKUP, + REPLICA_COUNT, + group(), + NoopBus, + LocalPipeline::new(), + ); + consensus.set_view(OLD_VIEW); + consensus.set_log_view(OLD_VIEW); + consensus.sequencer().set_sequence(LAGGING_HEAD); + consensus.restore_commit_state(LAGGING_HEAD - 1, LAGGING_HEAD - 1); + consensus.set_local_dvc_suffix(suffix_all_present( + (LAGGING_HEAD - 1..=LAGGING_HEAD) + .rev() + .map(prepare) + .collect(), + )); + consensus.set_recovered_log_complete(false); + consensus.init(); + consensus + }; + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, RECOVERED_HEAD, false), + lagging, + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + }; + + cluster.run(TICKS, |from, to| from != SURVIVOR && to != SURVIVOR); + + assert!(!cluster.replica(LAGGING_BACKUP).has_untrusted_log()); + assert!(cluster.replica(RESTARTED_PRIMARY).has_untrusted_log()); + for delivered in &cluster.do_view_changes { + assert_eq!( + delivered.log_untrusted, + delivered.from == RESTARTED_PRIMARY, + "replica {} sent a DoViewChange misreporting its log", + delivered.from + ); + } + assert!( + cluster + .do_view_changes + .iter() + .any(|delivered| delivered.from == RESTARTED_PRIMARY) + ); + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + } + + /// A trusted candidate counts its own `DoViewChange` as trusted: with one + /// trusted peer it has a quorum and merges without the untrusted one, even + /// though that sender reports the longest log. + #[test] + fn given_trusted_candidate_when_one_trusted_peer_reports_should_merge_without_the_untrusted_log() + { + let plane = PlaneKind::Partitions; + let candidate = { + let mut consensus = VsrConsensus::new( + 1, + LAGGING_BACKUP, + REPLICA_COUNT, + group(), + NoopBus, + LocalPipeline::new(), + ); + consensus.set_view(OLD_VIEW); + consensus.set_log_view(OLD_VIEW); + consensus.sequencer().set_sequence(RECOVERED_HEAD); + consensus.restore_commit_state(LAGGING_HEAD, LAGGING_HEAD); + consensus.set_local_dvc_suffix(suffix_all_present( + (LAGGING_HEAD..=RECOVERED_HEAD).rev().map(prepare).collect(), + )); + consensus.set_recovered_log_complete(false); + consensus.init(); + consensus + }; + let view = OLD_VIEW + 1; + assert_eq!(candidate.primary_index(view), LAGGING_BACKUP); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + candidate, + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + }; + let svc = VsrAction::SendStartViewChange { + view, + group: group(), + }; + cluster.receive(SURVIVOR, LAGGING_BACKUP, &svc); + let do_view_change = + |head: u64, commit: u64, log_untrusted: bool| VsrAction::SendDoViewChange { + view, + target: LAGGING_BACKUP, + log_view: OLD_VIEW, + op: head, + commit, + group: group(), + suffix: suffix_all_present((commit..=head).rev().map(prepare).collect()), + log_untrusted, + }; + + cluster.receive( + RESTARTED_PRIMARY, + LAGGING_BACKUP, + &do_view_change(COMMITTED, COMMITTED - 1, true), + ); + assert!( + !cluster.replica(LAGGING_BACKUP).view_log_is_pending(), + "an untrusted DoViewChange completed the candidate's quorum" + ); + + cluster.receive( + SURVIVOR, + LAGGING_BACKUP, + &do_view_change(RECOVERED_HEAD + 1, LAGGING_HEAD, false), + ); + assert_eq!( + cluster + .replica(LAGGING_BACKUP) + .with_pending_view_log(|log| log.op_head), + Some(RECOVERED_HEAD + 1), + "the merge took the untrusted sender's longer log" + ); + } } diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs index d85a4da4e..44141b3fb 100644 --- a/core/consensus/src/observability.rs +++ b/core/consensus/src/observability.rs @@ -230,6 +230,8 @@ pub struct ReplicaLogContext { pub commit: u64, pub status: Status, pub role: ReplicaRole, + /// See [`VsrConsensus::has_untrusted_log`]. + pub untrusted_log: bool, } impl ReplicaLogContext { @@ -255,6 +257,7 @@ impl ReplicaLogContext { commit: consensus.commit_max(), status: consensus.status(), role, + untrusted_log: consensus.has_untrusted_log(), } } } @@ -519,6 +522,7 @@ fn emit_partition_diag_error(event: &PartitionDiagEvent<'_>) { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, operation, client, request, @@ -557,6 +561,7 @@ fn emit_partition_diag_warn(event: &PartitionDiagEvent<'_>) { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, operation, client, request, @@ -595,6 +600,7 @@ fn emit_partition_diag_info(event: &PartitionDiagEvent<'_>) { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, operation, client, request, @@ -633,6 +639,7 @@ fn emit_partition_diag_debug(event: &PartitionDiagEvent<'_>) { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, operation, client, request, @@ -671,6 +678,7 @@ fn emit_partition_diag_trace(event: &PartitionDiagEvent<'_>) { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, operation, client, request, @@ -772,6 +780,7 @@ pub fn emit_replica_event(sim_event: SimEventKind, ctx: &ReplicaLogContext) { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, ); } @@ -797,6 +806,7 @@ pub fn emit_namespace_progress_event( commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, op, pipeline_depth, ); @@ -821,6 +831,7 @@ impl StructuredSimEvent for RequestLogEvent { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, client_id = self.client_id, request_id = self.request_id, operation = operation_as_str(self.operation), @@ -847,6 +858,7 @@ impl StructuredSimEvent for PrepareLogEvent { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, op = self.op, parent_checksum = self.parent_checksum, prepare_checksum = self.prepare_checksum, @@ -877,6 +889,7 @@ impl StructuredSimEvent for AckLogEvent { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, op = self.op, prepare_checksum = self.prepare_checksum, ack_from_replica = self.ack_from_replica, @@ -906,6 +919,7 @@ impl StructuredSimEvent for CommitLogEvent { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, op = self.op, client_id = self.client_id, request_id = self.request_id, @@ -936,6 +950,7 @@ impl StructuredSimEvent for ViewChangeLogEvent { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, old_view = self.old_view, new_view = self.new_view, reason = self.reason.as_str(), @@ -962,6 +977,7 @@ impl StructuredSimEvent for ControlActionLogEvent { commit = ctx.commit, status = status_as_str(ctx.status), role = ctx.role.as_str(), + untrusted_log = ctx.untrusted_log, action = self.action.as_str(), target_replica = self.target_replica.unwrap_or_default(), op = self.op.unwrap_or_default(), diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index d846c9914..448f33fbd 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -1105,7 +1105,8 @@ mod tests { commit, group: 0, log_view: 0, - reserved: [0; 68], + flags: 0, + reserved: [0; 67], nack_bitset: 0, present_bitset: 0, }; @@ -1503,7 +1504,8 @@ mod tests { commit, group: 0, log_view, - reserved: [0; 68], + flags: 0, + reserved: [0; 67], nack_bitset: 0, present_bitset: present, }; @@ -1538,7 +1540,8 @@ mod tests { commit, group: 0, log_view, - reserved: [0; 68], + flags: 0, + reserved: [0; 67], nack_bitset: 0, present_bitset: 0, }; diff --git a/core/consensus/src/view_change_quorum.rs b/core/consensus/src/view_change_quorum.rs index 89d5c266c..4a77f83e7 100644 --- a/core/consensus/src/view_change_quorum.rs +++ b/core/consensus/src/view_change_quorum.rs @@ -211,6 +211,9 @@ pub struct StoredDvc { /// The sender's uncommitted suffix. Empty from a silent sender, which counts /// toward the quorum and `max(commit)` but neither nacks nor offers bodies. pub suffix: DvcSuffix, + /// The sender's log may lack ops it acknowledged before a restart. See + /// `DoViewChangeHeader::flags` and `merge_view_change`. + pub log_untrusted: bool, } /// Array type for storing DVC messages from all replicas. diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index a6d425c4d..00fb4beea 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -1773,6 +1773,7 @@ impl Streams { topic_slab: usize, target_partitions: &[CreatedPartitionAssignment], created_view: u32, + durability: iggy_common::Durability, ) { let stream_wire = WireIdentifier::numeric(u32::try_from(stream_slab).expect("sim stream slab fits u32")); @@ -1801,8 +1802,8 @@ impl Streams { name: WireName::new(format!("sim-topic-{stream_slab}-{slab}")) .expect("sim topic name is valid"), options: iggy_common::TopicCreateOptions { - durability: iggy_common::Durability::Persisted, - consumer_offset_durability: iggy_common::Durability::Persisted, + durability, + consumer_offset_durability: durability, ..Default::default() } .to_wire() @@ -1844,6 +1845,27 @@ impl Streams { namespace: IggyNamespace, consensus_group_id: u64, created_view: u32, + ) { + self.seed_namespace_with_durability( + namespace, + consensus_group_id, + created_view, + iggy_common::Durability::Persisted, + ); + } + + /// [`Self::seed_namespace`] with the topic's message and consumer-offset + /// durability set to `durability`. Only a topic the call creates takes it. + /// + /// # Panics + /// As [`Self::seed_namespace`]. + #[cfg(any(test, feature = "simulator"))] + pub fn seed_namespace_with_durability( + &self, + namespace: IggyNamespace, + consensus_group_id: u64, + created_view: u32, + durability: iggy_common::Durability, ) { let stream_slab = namespace.stream_id(); let topic_slab = namespace.topic_id(); @@ -1877,7 +1899,13 @@ impl Streams { }) .collect(); self.seed_stream_slabs(stream_slab); - self.seed_topic_slabs(stream_slab, topic_slab, &target_partitions, created_view); + self.seed_topic_slabs( + stream_slab, + topic_slab, + &target_partitions, + created_view, + durability, + ); if self.created_revision_for_namespace(namespace).is_some() { return; diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index dd85ade26..11a1020ee 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -919,6 +919,9 @@ where partition.log.journal().inner.set_repair_retention(false); } partition + .consensus + .set_recovered_log_complete(partition.acknowledges_only_written_prepares()); + partition } /// Record the metadata incarnation this partition was built for. @@ -2676,6 +2679,17 @@ where self.offset_space.append_live } + /// The frontier a pending quarantine records for the rebuild, if one is + /// pending. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub const fn quarantine_frontier(&self) -> Option<u64> { + match &self.transition { + Some(PendingPartitionTransition::Quarantine { frontier, .. }) => Some(*frontier), + _ => None, + } + } + /// Capture the ideal-storage simulator's durable state at process loss. #[cfg(any(test, feature = "simulator"))] pub fn take_retained_state(&mut self) -> crate::RetainedPartitionState { @@ -2909,6 +2923,23 @@ where } } + /// [`Self::offset_frontier`] as it stood once every op below `op` was + /// applied, read off the stamped entries this replica still holds. + /// + /// `None` when an applied op from `op` up to the first one that minted + /// offsets is no longer retained, so the frontier there cannot be known. + #[must_use] + pub fn offset_frontier_before(&self, op: u64) -> Option<u64> { + let journal = &self.log.journal().inner; + for applied in op..=self.consensus.commit_min() { + let entry = journal.repair_entry(applied)?; + if let Some(batch) = materialization_batch_meta(&entry, None).ok()? { + return Some(batch.base_offset); + } + } + Some(self.offset_frontier()) + } + /// The offset the next mint will take, which is where the segment chain has /// to be anchored for the appends that follow to land contiguously. /// @@ -3305,8 +3336,17 @@ where /// Install this topic's runtime knobs, as resolved at topic admission. /// Unset fields keep the shard-wide configured values. - pub const fn set_runtime_options(&mut self, runtime_options: TopicRuntimeOptions) { + pub fn set_runtime_options(&mut self, runtime_options: TopicRuntimeOptions) { self.runtime_options = runtime_options; + self.consensus + .set_recovered_log_complete(self.acknowledges_only_written_prepares()); + } + + /// Whether every prepare this replica acknowledges is on stable storage + /// first, so a restart recovers all of them. `replicated` acknowledges once + /// the write is queued, and either policy covers some of the log. + const fn acknowledges_only_written_prepares(&self) -> bool { + self.durability().is_persisted() && self.consumer_offset_durability().is_persisted() } #[must_use] @@ -6344,6 +6384,15 @@ where self.workerless_offset_files.clear(); } + /// Checksum of the prepare accepted into this replica's WAL at `op`, readable after + /// the journal has evicted the header and until a checkpoint reclaims it. + #[must_use] + pub fn wal_checksum(&self, op: u64) -> Option<u128> { + self.persistence + .as_ref() + .and_then(|persistence| persistence.checksum(op)) + } + /// The commit fault that fenced this partition, if one has. #[must_use] pub const fn fatal(&self) -> Option<&FatalCommit> { @@ -14045,6 +14094,33 @@ mod tests { assert_eq!(persistence.checkpoint_op(), 2); } + #[test] + fn given_topic_durability_when_installed_should_tell_consensus_whether_restarts_keep_acks() { + use iggy_common::Durability::{Persisted, Replicated}; + let mut partition = test_partition(); + assert!( + !partition.consensus().recovered_log_complete(), + "the default `replicated` policy acknowledges before the write" + ); + for (durability, consumer_offset_durability, complete) in [ + (Persisted, Persisted, true), + (Replicated, Persisted, false), + (Persisted, Replicated, false), + (Replicated, Replicated, false), + ] { + partition.set_runtime_options(TopicRuntimeOptions { + durability, + consumer_offset_durability, + ..TopicRuntimeOptions::default() + }); + assert_eq!( + partition.consensus().recovered_log_complete(), + complete, + "{durability:?} messages, {consumer_offset_durability:?} offsets" + ); + } + } + pub(super) fn test_partition() -> IggyPartition<IggyMessageBus> { let namespace = IggyNamespace::new(1, 1, 0); let consensus = VsrConsensus::new( diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 3a4d54ef3..3e766cbfb 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -4696,7 +4696,8 @@ where #[allow(clippy::future_not_send, clippy::too_many_lines)] async fn on_start_view(&self, msg: Message<StartViewHeader>) where - B: MessageBus, + B: MessageBus + 'static, + T: ShardsTable, MJ: JournalHandle, <MJ as JournalHandle>::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, @@ -4837,6 +4838,18 @@ where // like it, pending-less adoptions (empty StartView suffix) still sweep // the relics above the adopted head. let pending = partition.consensus().pending_view_log(); + if let Some((diverged_op, agreed_op)) = pending + .as_ref() + .and_then(|pending| partition_applied_divergence(partition, pending)) + { + self.resync_diverged_partition( + IggyNamespace::from_raw(header.group), + partition, + diverged_op, + agreed_op, + ); + return; + } if partition.has_io_dispatcher() && let Some(from_op) = partition_view_divergence_from(self.id, partition, pending.as_ref()) @@ -5920,7 +5933,8 @@ where #[allow(clippy::future_not_send)] async fn advance_pending_partition_view(&self, namespace: IggyNamespace) where - B: MessageBus, + B: MessageBus + 'static, + T: ShardsTable, MJ: JournalHandle, <MJ as JournalHandle>::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, @@ -5942,6 +5956,12 @@ where let Some(pending) = partition.consensus().pending_view_log() else { return; }; + if let Some((diverged_op, agreed_op)) = + partition_applied_divergence(partition, &pending) + { + self.resync_diverged_partition(namespace, partition, diverged_op, agreed_op); + return; + } // Before the scan can mean anything: the repair ingest skips an op it // already holds a header for, so the scan would report a gap nothing // fills. Backups reach this on StartView adoption; a primary-elect has no @@ -7631,7 +7651,8 @@ where namespace_scratch: &mut Vec<IggyNamespace>, ) -> Option<FatalCommit> where - B: MessageBus, + B: MessageBus + 'static, + T: ShardsTable, MJ: JournalHandle, <MJ as JournalHandle>::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, @@ -7676,9 +7697,11 @@ where let mut persistence_metrics = partitions::PersistenceMetrics::default(); let mut repair_ring_entries = 0usize; let mut repair_ring_bytes = 0u64; + let mut untrusted_logs = 0usize; for namespace in namespace_scratch.iter() { if let Some(partition) = partitions.get_mut_by_ns(namespace) { partition.drive_persistence().await; + untrusted_logs += usize::from(partition.consensus().has_untrusted_log()); let (entries, bytes) = partition.repair_ring_occupancy(); repair_ring_entries += entries; repair_ring_bytes += bytes; @@ -7702,6 +7725,7 @@ where .record_replica_reads(&self.bus.take_replica_read_stats()); self.metrics .set_repair_ring(repair_ring_entries, repair_ring_bytes); + self.metrics.set_partitions_untrusted_log(untrusted_logs); // Counted at most ONCE per sweep and only if a re-arm actually fires, // then tracked locally as arms land. Counting per namespace is a full @@ -8833,6 +8857,60 @@ where false } + /// Replace a partition whose applied history the view contradicts with a + /// fresh copy from its peers, instead of letting the WAL's parent check + /// stop the whole node on the view's next prepare. + /// + /// The rebuild quarantines the local segments as evidence and comes back + /// with no materialization, so it serves, acks and votes nothing until a + /// state transfer from a settled primary installs the view's history. + /// + /// `agreed_op` is the highest applied op the view confirms. The recorded + /// frontier may sit below the group's but never above it: too high refuses + /// every offer on a topic nobody writes to. Falling back to `0` is safe + /// because the quarantined partition holds no data to rewind, and an offer + /// from a lagging peer is still a consistent prefix of the agreed history. + fn resync_diverged_partition( + &self, + namespace: IggyNamespace, + partition: &mut IggyPartition<B, SB>, + diverged_op: u64, + agreed_op: Option<u64>, + ) where + B: MessageBus + 'static, + T: ShardsTable, + { + tracing::error!( + shard = self.id, + namespace_raw = namespace.inner(), + diverged_op, + view = partition.consensus().view(), + commit_min = partition.consensus().commit_min(), + "the view contradicts applied partition history at op {diverged_op}; \ + quarantining this partition and resyncing it by state transfer" + ); + self.metrics.record_partition_applied_divergence(); + // The rebuild refuses any offer below the frontier it records, and the + // view's history past `agreed_op` can be shorter than this replica's. + // Counting this replica's own divergent mints would refuse every offer + // until the group wrote past them, which an idle topic never does. + let frontier = agreed_op + .and_then(|agreed| partition.offset_frontier_before(agreed.saturating_add(1))) + .unwrap_or_else(|| { + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + diverged_op, + ?agreed_op, + "no retained applied op is confirmed by the view below op {diverged_op}; \ + the rebuild records no offset frontier and accepts any offer" + ); + 0 + }); + self.drop_partition_transfer_state(namespace, partition); + self.fence_partition_for_rebuild(namespace, partition, frontier); + } + /// Fence one partition for rebuild: quarantine its segment files, drop it /// from routing, and queue the retirement the reconciler re-materialises /// from committed metadata. @@ -11258,6 +11336,99 @@ fn build_dvc_suffix( DvcSuffix::new(headers, nack_bitset, present_bitset) } +/// An applied op whose history the view contradicts, or `None` when the view +/// extends what this replica applied. +/// +/// Truncation stops above the applied floor, so a contradiction at or below it +/// is not repairable in place: the WAL refuses the view's next prepare because it +/// names a parent this replica never wrote, and that refusal used to stop the +/// node. Checked two ways, because the journal evicts applied headers: +/// * a canonical header at or below the floor against the local checksum there, +/// * the canonical header just above the floor, whose `parent` names the +/// checksum the view holds at the floor. +/// +/// An unsealed checksum on either side is not evidence, as in +/// [`header_is_view_entry`]. +fn applied_history_divergence( + pending: &MergedLog, + applied_floor: u64, + checksum_at: impl Fn(u64) -> Option<u128>, +) -> Option<u64> { + let contradicts = |op: u64, canonical: u128| { + checksum_at(op).is_some_and(|local| { + local != CHECKSUM_UNSEALED && canonical != CHECKSUM_UNSEALED && local != canonical + }) + }; + pending.headers.iter().find_map(|canonical| { + if canonical.op <= applied_floor { + contradicts(canonical.op, canonical.checksum).then_some(canonical.op) + } else if applied_floor > 0 && applied_floor.checked_add(1) == Some(canonical.op) { + contradicts(applied_floor, canonical.parent).then_some(applied_floor) + } else { + None + } + }) +} + +/// The highest applied op below `diverged_op` whose history the view confirms, +/// or `None` when no header the view carries vouches for one. +/// +/// Checksums chain through `parent`, so a match at an op confirms every op +/// below it. The pending headers run newest first and the divergence check +/// reports the newest contradiction, which can sit above older ones this +/// replica also applied, so ops at and above it never count. Checked the same +/// two ways as [`applied_history_divergence`]. +fn applied_history_agreement( + pending: &MergedLog, + applied_floor: u64, + diverged_op: u64, + checksum_at: impl Fn(u64) -> Option<u128>, +) -> Option<u64> { + let confirms = |op: u64, canonical: u128| { + op <= applied_floor + && op < diverged_op + && canonical != CHECKSUM_UNSEALED + && checksum_at(op).is_some_and(|local| local == canonical) + }; + pending + .headers + .iter() + .filter_map(|canonical| { + let direct = confirms(canonical.op, canonical.checksum).then_some(canonical.op); + let via_parent = canonical + .op + .checked_sub(1) + .filter(|&parent_op| confirms(parent_op, canonical.parent)); + direct.max(via_parent) + }) + .max() +} + +/// [`applied_history_divergence`] against this partition's journal and WAL, +/// paired with the [`applied_history_agreement`] below it. +fn partition_applied_divergence<B, SB>( + partition: &IggyPartition<B, SB>, + pending: &MergedLog, +) -> Option<(u64, Option<u64>)> +where + B: MessageBus, + SB: SuperblockStore, +{ + let applied_floor = partition.consensus().commit_min(); + let checksum_at = |op| { + partition + .log + .journal() + .inner + .header_by_op(op) + .map(|header| header.checksum) + .or_else(|| partition.wal_checksum(op)) + }; + let diverged_op = applied_history_divergence(pending, applied_floor, checksum_at)?; + let agreed_op = applied_history_agreement(pending, applied_floor, diverged_op, checksum_at); + Some((diverged_op, agreed_op)) +} + /// Partition-plane twin of `Shard::reconcile_metadata_view_divergence`: same split /// at the announced commit point, dropping above it and reporting at or below. /// @@ -11561,6 +11732,7 @@ async fn dispatch_vsr_actions<B, P, J>( commit, group, suffix, + log_untrusted, } => { let header_size = size_of::<DoViewChangeHeader>(); let total_size = header_size + suffix.encoded_len(); @@ -11580,6 +11752,9 @@ async fn dispatch_vsr_actions<B, P, J>( h.op = *op; h.commit = *commit; h.group = *group; + if *log_untrusted { + h.flags |= DoViewChangeHeader::FLAG_LOG_UNTRUSTED; + } h.nack_bitset = nack_bitset; h.present_bitset = present_bitset; h.checksum_body = body_checksum; @@ -11767,12 +11942,19 @@ async fn dispatch_partition_wire_actions<B, P, J, SB>( return; } if partition.requires_state_transfer() { - actions.retain(|action| matches!(action, VsrAction::SendRequestStartView { .. })); + actions.retain(sendable_while_awaiting_transfer); } dispatch_vsr_actions::<B, P, J>(consensus, None, &actions).await; dispatch_partition_journal_actions(consensus, partition, &actions).await; } +/// Wire actions a partition awaiting a state transfer may still send: its +/// probe, and the empty vote that vouches for no history but lets a group +/// whose every log is untrusted start a view. +fn sendable_while_awaiting_transfer(action: &VsrAction) -> bool { + matches!(action, VsrAction::SendRequestStartView { .. }) || action.is_empty_vote() +} + #[allow( clippy::future_not_send, clippy::too_many_lines, @@ -12054,7 +12236,10 @@ mod repair_scope_tests { mod view_coverage_tests { //! Holding an op is not holding the view's op. - use super::{MergedLog, ScanFloor, first_op_not_covered}; + use super::{ + MergedLog, ScanFloor, applied_history_agreement, applied_history_divergence, + first_op_not_covered, + }; use iggy_binary_protocol::{Command, Operation, PrepareHeader}; /// What a caught-up replica passes: nothing compacted, nothing lagging. @@ -12077,6 +12262,153 @@ mod view_coverage_tests { header } + /// A view whose log runs `from..=to` over `request`, hash-chained. + fn view_log(from: u64, to: u64, request: u64) -> MergedLog { + let mut parent = sealed(from - 1, request).checksum; + let mut headers = Vec::new(); + for op in from..=to { + let mut header = sealed(op, request); + header.parent = parent; + header.checksum = header.identity_checksum(); + parent = header.checksum; + headers.push(header); + } + headers.reverse(); + MergedLog { + op_head: to, + commit_max: from, + headers, + committed_elsewhere: Vec::new(), + } + } + + #[test] + fn given_view_contradicting_an_applied_op_when_checked_should_report_it() { + // Applied through 5 from one history; the view's op 5 is another. Only a + // state transfer can replace an applied op. The view stops at 5, so no + // header above the floor can give it away through its parent. + let pending = view_log(4, 5, 1); + let applied = sealed(5, 9).checksum; + assert_eq!( + applied_history_divergence(&pending, 5, |op| (op == 5).then_some(applied)), + Some(5) + ); + } + + #[test] + fn given_evicted_applied_head_when_view_names_another_parent_should_report_it() { + // The journal evicted everything applied, so the view's first op above the + // floor is the only witness: its parent is not the checksum this replica + // wrote at op 4, so the WAL would refuse it. + let pending = view_log(5, 7, 1); + let applied_four = sealed(4, 9).checksum; + assert_eq!( + applied_history_divergence(&pending, 4, |op| (op == 4).then_some(applied_four)), + Some(4) + ); + } + + #[test] + fn given_applied_floor_zero_when_view_names_another_parent_should_report_nothing() { + // Nothing is applied, so op 0 is no history this replica could lose. + let pending = view_log(1, 3, 1); + let local_zero = sealed(0, 9).checksum; + assert_eq!( + applied_history_divergence(&pending, 0, |op| (op == 0).then_some(local_zero)), + None + ); + } + + #[test] + fn given_divergent_applied_suffix_when_measuring_agreement_should_stop_below_it() { + // Ops 1..=3 match the view, ops 4 and 5 were minted by this replica alone. + // The divergence check reports op 5 first, so anything at or above op 4 + // counted as agreed would rebuild with a frontier the group never reached. + let view = view_log(1, 5, 1); + let local_suffix = view_log(4, 5, 9); + let local: Vec<(u64, u128)> = view + .headers + .iter() + .filter(|header| header.op <= 3) + .chain(&local_suffix.headers) + .map(|header| (header.op, header.checksum)) + .collect(); + let checksum_at = |op: u64| { + local + .iter() + .find(|(local_op, _)| *local_op == op) + .map(|(_, checksum)| *checksum) + }; + let window = |ops: &[u64]| MergedLog { + headers: view + .headers + .iter() + .filter(|header| ops.contains(&header.op)) + .copied() + .collect(), + ..view.clone() + }; + + let with_agreed_op = window(&[5, 4, 3]); + assert_eq!( + applied_history_divergence(&with_agreed_op, 5, checksum_at), + Some(5) + ); + assert_eq!( + applied_history_agreement(&with_agreed_op, 5, 5, checksum_at), + Some(3) + ); + + let via_parent = window(&[5, 4]); + assert_eq!( + applied_history_agreement(&via_parent, 5, 5, checksum_at), + Some(3) + ); + + let no_witness = window(&[5]); + assert_eq!( + applied_history_agreement(&no_witness, 5, 5, checksum_at), + None + ); + } + + #[test] + fn given_view_extending_applied_history_when_checked_should_report_nothing() { + let pending = view_log(4, 7, 1); + let applied: Vec<_> = pending + .headers + .iter() + .map(|header| (header.op, header.checksum)) + .collect(); + let checksum_at = |op: u64| { + applied + .iter() + .find(|(applied_op, _)| *applied_op == op) + .map(|(_, checksum)| *checksum) + }; + assert_eq!(applied_history_divergence(&pending, 4, checksum_at), None); + assert_eq!(applied_history_divergence(&pending, 5, checksum_at), None); + assert_eq!( + applied_history_divergence(&pending, 4, |_| None), + None, + "an unknown local checksum is not evidence" + ); + assert_eq!( + applied_history_divergence(&pending, 4, |_| Some(super::CHECKSUM_UNSEALED)), + None, + "an unsealed local checksum is not evidence" + ); + let mut unsealed = pending; + for header in &mut unsealed.headers { + header.checksum = super::CHECKSUM_UNSEALED; + } + assert_eq!( + applied_history_divergence(&unsealed, 5, checksum_at), + None, + "an unsealed canonical checksum is not evidence" + ); + } + #[test] fn given_a_diverging_entry_when_scanning_should_report_it_like_a_hole() { // Op 99 is present and is not the view's op 99. Reading presence as coverage @@ -13578,3 +13910,96 @@ mod partition_ack_durability_tests { } } } + +#[cfg(test)] +mod untrusted_do_view_change_tests { + use super::*; + use consensus::{DvcSuffix, LocalPipeline}; + use iggy_common::{IggyByteSize, PartitionStats}; + use journal::prepare_journal::PrepareJournal; + use message_bus::IggyMessageBus; + use server_common::iobuf::Owned; + use std::sync::Arc; + + const GROUP: u64 = 42; + + fn do_view_change(op: u64, log_untrusted: bool) -> VsrAction { + VsrAction::SendDoViewChange { + view: 1, + target: 1, + log_view: 0, + op, + commit: 0, + group: GROUP, + suffix: DvcSuffix::empty(), + log_untrusted, + } + } + + #[compio::test] + async fn given_untrusted_log_when_dispatching_do_view_change_should_stamp_the_flag() { + let bus = IggyMessageBus::new(0); + let sent = Rc::new(RefCell::new(Vec::new())); + let captured = sent.clone(); + bus.set_replica_forward_fn(Box::new(move |_, _, frame| { + captured.borrow_mut().push(frame); + Ok(()) + })); + for replica in 1..3 { + assert!(bus.owner_table().try_claim(replica, 1)); + } + let consensus = VsrConsensus::new(1, 0, 3, GROUP, bus, LocalPipeline::new()); + consensus.init(); + let partition: Box<IggyPartition<IggyMessageBus>> = + Box::new(IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + IggyByteSize::from(1024 * 1024), + )); + + for log_untrusted in [true, false] { + sent.borrow_mut().clear(); + dispatch_partition_wire_actions::<_, _, PrepareJournal, _>( + partition.consensus(), + &partition, + vec![do_view_change(0, log_untrusted)], + ) + .await; + let frames = sent.borrow(); + assert_eq!(frames.len(), 2, "a DoViewChange is broadcast"); + for frame in frames.iter() { + let message = Message::<DoViewChangeHeader>::try_from(Owned::copy_from_slice( + frame.as_slice(), + )) + .expect("a dispatched DoViewChange decodes"); + assert_eq!(message.header().log_untrusted(), log_untrusted); + assert_eq!(message.header().verify_frame(), Ok(())); + } + } + } + + #[test] + fn given_partition_awaiting_transfer_should_send_only_its_probe_and_empty_vote() { + assert!(sendable_while_awaiting_transfer( + &VsrAction::SendRequestStartView { + view: 1, + group: GROUP, + } + )); + assert!(sendable_while_awaiting_transfer(&do_view_change(0, true))); + assert!( + !sendable_while_awaiting_transfer(&do_view_change(0, false)), + "a trusted vote vouches for a log the partition does not have" + ); + assert!( + !sendable_while_awaiting_transfer(&do_view_change(5, true)), + "a vote carrying a head vouches for history" + ); + assert!(!sendable_while_awaiting_transfer( + &VsrAction::SendStartViewChange { + view: 1, + group: GROUP, + } + )); + } +} diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index d544b2154..93ea36b48 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -244,6 +244,8 @@ pub struct ShardMetrics { partitions_reconcile_failures_total: Counter, partitions_duplicate_builds_discarded_total: Counter, partition_transfer_refusals_total: Counter, + partition_applied_divergence_total: Counter, + partitions_untrusted_log: Gauge, partition_frames_rejected_stale_total: Counter, partition_frames_rejected_ahead_total: Counter, partition_requests_denied_transient_total: Counter, @@ -320,6 +322,8 @@ impl ShardMetrics { partitions_reconcile_failures_total: Counter::default(), partitions_duplicate_builds_discarded_total: Counter::default(), partition_transfer_refusals_total: Counter::default(), + partition_applied_divergence_total: Counter::default(), + partitions_untrusted_log: Gauge::default(), partition_frames_rejected_stale_total: Counter::default(), partition_frames_rejected_ahead_total: Counter::default(), partition_requests_denied_transient_total: Counter::default(), @@ -606,6 +610,38 @@ impl ShardMetrics { self.partition_transfer_refusals_total.get() } + /// Bumped when a view this replica adopts contradicts a partition op it + /// already applied, and the partition is set aside for a resync. + /// + /// Any non-zero value means committed history diverged between replicas: + /// the resync keeps the node serving, but the local copy it quarantined + /// into `.fenced.N` holds an op the group did not keep. + pub fn record_partition_applied_divergence(&self) { + self.partition_applied_divergence_total.inc(); + } + + /// Test-only read, mirroring the siblings. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn partition_applied_divergence_value(&self) -> u64 { + self.partition_applied_divergence_total.get() + } + + /// Republished by every partition sweep: how many of this shard's + /// partitions elected on a log that may lack acknowledged ops and are + /// waiting for every replica to report before they serve again. + pub fn set_partitions_untrusted_log(&self, count: usize) { + self.partitions_untrusted_log + .set(i64::try_from(count).unwrap_or(i64::MAX)); + } + + /// Test-only read, mirroring the siblings. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn partitions_untrusted_log_value(&self) -> i64 { + self.partitions_untrusted_log.get() + } + /// Bumped when a parked partition frame is answered instead of served /// because it was addressed to an incarnation this shard no longer holds /// (delete + recreate recycled the namespace's slab keys). Serving it would @@ -884,6 +920,18 @@ impl ShardMetrics { "partition state transfers refused by the serving peer", self.partition_transfer_refusals_total.clone(), ); + registry.register( + "partition_applied_divergence", + "adopted views that contradicted an applied partition op; non-zero means committed \ + history diverged", + self.partition_applied_divergence_total.clone(), + ); + registry.register( + "partitions_untrusted_log", + "partitions electing on a log that may lack acknowledged ops; a view starts only \ + from a quorum of trusted logs or with every replica reporting", + self.partitions_untrusted_log.clone(), + ); registry.register( "partition_frames_rejected_stale", "parked frames rejected for a dead incarnation; non-zero is a caught anomaly", diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 0e57c85eb..38d3a1cff 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -37,7 +37,7 @@ use deps::{MemStorage, SimJournal}; use executor::{DetExecutor, RunOutcome, TaskId}; use iggy_binary_protocol::primitives::partition_history::PartitionContext; use iggy_binary_protocol::{Command, GenericHeader, PrepareHeader, ReplyHeader}; -use iggy_common::IggyError; +use iggy_common::{Durability, IggyError}; use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind}; use metadata::impls::metadata::StreamsFrontend; use network::Network; @@ -79,7 +79,7 @@ const SETUP_TOTAL_STEPS: u32 = 4_000; pub struct SimReplica { /// Fixture metadata this replica observed before any WAL replay. Runtime /// allocations must be recovered from their logged commands instead. - seeded_namespaces: RefCell<BTreeMap<IggyNamespace, u32>>, + seeded_namespaces: RefCell<BTreeMap<IggyNamespace, (u32, Durability)>>, /// Shards of this replica, indexed by shard id. pub shards: Vec<Rc<Replica>>, /// Shard 0's durable superblock. Held here, not in the shard, so its bytes @@ -228,6 +228,9 @@ pub struct Simulator { /// of that namespace, a restart's included, so no replica seeds a view its /// peers did not. partition_created_views: HashMap<IggyNamespace, u32>, + /// Topic durability a namespace is seeded with, `persisted` unless a test + /// chose otherwise through [`Self::set_partition_durability`]. + partition_durabilities: HashMap<IggyNamespace, Durability>, /// Replies a setup handshake pulled off the wire that were not its own. /// /// `await_setup_reply` steps the whole simulator, so it sees every client's @@ -542,6 +545,7 @@ impl Simulator { entry_rng: Xoshiro256PlusPlus::seed_from_u64(SimSeeds::derive(seed).entry_shard), evicted: Vec::new(), partition_created_views: HashMap::new(), + partition_durabilities: HashMap::new(), deferred_client_replies: Vec::new(), seed, shell, @@ -602,6 +606,7 @@ impl Simulator { let Some(created_view) = self.partition_created_view(namespace) else { return; }; + let durability = self.partition_durability(namespace); for (i, replica) in self.replicas.iter().enumerate() { if self.crashed.contains(&(i as u8)) { continue; @@ -611,8 +616,13 @@ impl Simulator { replica .seeded_namespaces .borrow_mut() - .insert(namespace, created_view); - streams.seed_namespace(namespace, namespace.inner(), created_view); + .insert(namespace, (created_view, durability)); + streams.seed_namespace_with_durability( + namespace, + namespace.inner(), + created_view, + durability, + ); } materialise_partition(replica, namespace, self.consumer_offsets_max); } @@ -633,6 +643,7 @@ impl Simulator { let Some(created_view) = self.partition_created_view(namespace) else { return; }; + let durability = self.partition_durability(namespace); for (i, replica) in self.replicas.iter().enumerate() { if self.crashed.contains(&(i as u8)) { continue; @@ -640,7 +651,7 @@ impl Simulator { replica .seeded_namespaces .borrow_mut() - .insert(namespace, created_view); + .insert(namespace, (created_view, durability)); // Shard 0 is the sole metadata writer. Peers see the seed through the // left-right publish, so seeding a reader-mode peer STM would panic. replica.shards[0] @@ -648,10 +659,32 @@ impl Simulator { .metadata() .mux_stm .streams() - .seed_namespace(namespace, namespace.inner(), created_view); + .seed_namespace_with_durability( + namespace, + namespace.inner(), + created_view, + durability, + ); } } + /// Seed `namespace`'s topic with `durability` for both messages and consumer + /// offsets. Call before the first seed; a topic already seeded keeps its own. + pub fn set_partition_durability(&mut self, namespace: IggyNamespace, durability: Durability) { + debug_assert!( + !self.partition_created_views.contains_key(&namespace), + "{namespace:?} is already seeded with its own durability" + ); + self.partition_durabilities.insert(namespace, durability); + } + + fn partition_durability(&self, namespace: IggyNamespace) -> Durability { + self.partition_durabilities + .get(&namespace) + .copied() + .unwrap_or(Durability::Persisted) + } + /// The view `namespace` was created in: the metadata plane's view at its /// first seed, as the prepare of a real create carries it, and the same value /// for every seed after. `None` with no live metadata consensus to read. @@ -1296,8 +1329,18 @@ impl Simulator { /// # Panics /// If the replica is not crashed, or its shard count does not fit `u16`; mesh /// construction caps it. - #[allow(clippy::too_many_lines)] pub fn replica_restart(&mut self, replica_index: u8) { + self.replica_restart_losing_partitions(replica_index, &[]); + } + + /// Restart a crashed replica whose `lost` partitions come back with no log, + /// as a server whose partition directories were wiped does: the group + /// re-materialises from its recorded view alone and may not elect on it. + /// + /// # Panics + /// As [`Self::replica_restart`]. + #[allow(clippy::too_many_lines)] + pub fn replica_restart_losing_partitions(&mut self, replica_index: u8, lost: &[IggyNamespace]) { assert!( self.crashed.contains(&replica_index), "cannot restart replica {replica_index}: not crashed" @@ -1321,11 +1364,14 @@ impl Simulator { // server's messages are in segment files its boot recovers the offset counter // from, so rebuilding with nothing would model total data loss rather than a // restart. Before the rebuild, which drops the shards. - let partition_logs = self.retain_partition_logs(idx, &partition_superblocks); + let mut partition_logs = self.retain_partition_logs(idx, &partition_superblocks); + for namespace in lost { + partition_logs.remove(namespace); + } let seeded_namespaces = self.replicas[idx].seeded_namespaces.borrow().clone(); let seed_namespaces: Vec<_> = seeded_namespaces .iter() - .map(|(&namespace, &view)| (namespace, view)) + .map(|(&namespace, &(view, durability))| (namespace, view, durability)) .collect(); let mut materialized_namespaces: Vec<_> = partition_superblocks.keys().copied().collect(); materialized_namespaces.sort_unstable(); @@ -3820,6 +3866,177 @@ mod tests { } } + /// Each replica's count of partitions waiting on every replica to report. + fn untrusted_log_gauges(sim: &Simulator, namespace: IggyNamespace) -> Vec<i64> { + sim.replicas + .iter() + .map(|replica| { + replica + .partition_shard(namespace) + .metrics() + .partitions_untrusted_log_value() + }) + .collect() + } + + /// Commit a send on a fresh three-replica partition, crash `crashed`, then + /// restart them. Every replica must settle on a log that still holds the + /// acknowledged send, trusting it again, and a new send must commit. + fn restart_keeps_acknowledged_send(durability: iggy_common::Durability, crashed: &[u8]) { + restart_keeps_acknowledged_send_losing(durability, crashed, None); + } + + /// [`restart_keeps_acknowledged_send`], with replica `lost` restarting + /// without the partition's log, so it may not elect on it. + fn restart_keeps_acknowledged_send_losing( + durability: iggy_common::Durability, + crashed: &[u8], + lost: Option<u8>, + ) { + const PROGRESS_STEPS: usize = 4_000; + const PAYLOAD: &[u8] = b"acknowledged-before-the-crash"; + let (mut sim, client) = cluster(packet::PacketSimulatorOptions::default().seed); + sim.register_client_with_primary(&client); + let namespace = IggyNamespace::new(1, 1, 0); + sim.set_partition_durability(namespace, durability); + sim.seed_stream_topic_partition(namespace); + for replica in &sim.replicas { + materialise_partition(replica, namespace, sim.consumer_offsets_max); + } + let primary = (0..PROGRESS_STEPS) + .find_map(|_| { + sim.step(); + (0..3).find(|replica| { + sim.partition_consensus_state(usize::from(*replica), namespace) + .is_some_and(|state| state.status == Status::Normal && state.is_primary) + }) + }) + .expect("the fresh partition elects a primary"); + sim.discover_partition_contexts(&client); + let request = client.send_messages(namespace, &[Bytes::from_static(PAYLOAD)]); + let committed = submit_and_wait_for_reply(&mut sim, client.client_id(), primary, request); + assert_eq!(committed.header().status, 0); + let committed_op = committed.header().commit; + let expected = sim + .offsets(usize::from(primary), namespace) + .expect("acknowledged partition offsets"); + + for replica in crashed { + sim.replica_crash(*replica); + } + for replica in crashed { + if lost == Some(*replica) { + sim.replica_restart_losing_partitions(*replica, &[namespace]); + } else { + sim.replica_restart(*replica); + } + } + + let mut gauge_saw_halt = false; + let settled = |sim: &Simulator| { + (0..3).all(|replica| { + sim.partition_consensus_state(replica, namespace) + .is_some_and(|state| { + state.status == Status::Normal && state.commit_min >= committed_op + }) + && sim.replicas[replica] + .partition_shard(namespace) + .plane + .partitions() + .get_by_ns(&namespace) + .is_some_and(|partition| !partition.consensus().has_untrusted_log()) + && sim.offsets(replica, namespace) == Some(expected) + }) + }; + assert!( + (0..PROGRESS_STEPS).any(|_| { + sim.step(); + gauge_saw_halt |= untrusted_log_gauges(&sim, namespace) + .iter() + .any(|gauge| *gauge > 0); + settled(&sim) + }), + "{durability:?}: replicas {crashed:?} restarted and the group never settled on the \ + acknowledged prefix: states={:?}, offsets={:?}", + (0..3) + .map(|replica| sim.partition_consensus_state(replica, namespace)) + .collect::<Vec<_>>(), + (0..3) + .map(|replica| sim.offsets(replica, namespace)) + .collect::<Vec<_>>() + ); + for replica in 0..3 { + let prepare = retained_prepare(&sim, replica, namespace, committed_op); + assert!( + prepare + .body() + .windows(PAYLOAD.len()) + .any(|bytes| bytes == PAYLOAD), + "replica {replica} lost the acknowledged payload at op {committed_op}" + ); + } + assert_eq!( + gauge_saw_halt, + durability == iggy_common::Durability::Replicated, + "{durability:?}: only a restarted replicated log waits on every replica" + ); + for _ in 0..PROGRESS_STEPS / 10 { + sim.step(); + } + assert_eq!(untrusted_log_gauges(&sim, namespace), vec![0; 3]); + + let primary = (0..3) + .find(|replica| { + sim.partition_consensus_state(usize::from(*replica), namespace) + .is_some_and(|state| state.is_primary) + }) + .expect("a settled group has a primary"); + sim.discover_partition_contexts(&client); + let request = client.send_messages(namespace, &[Bytes::from_static(b"after-restart")]); + let reply = submit_and_wait_for_reply(&mut sim, client.client_id(), primary, request); + assert_eq!(reply.header().status, 0); + } + + /// The double crash of #4476: the primary and one backup restart while the + /// third replica keeps running. Under `replicated` the restarted pair may + /// not elect past the survivor and the group settles once it reports; + /// `persisted` keeps the fast fallback. + #[test] + fn given_primary_and_backup_crash_when_they_restart_should_settle_and_accept_sends() { + for durability in [ + iggy_common::Durability::Replicated, + iggy_common::Durability::Persisted, + ] { + restart_keeps_acknowledged_send(durability, &[0, 1]); + } + } + + /// Every replica crashes and restarts. Nobody can answer a view probe, so + /// the fallback election must still settle the group, on the full log. + #[test] + fn given_every_replica_restarts_when_none_answers_a_probe_should_settle_and_accept_sends() { + for durability in [ + iggy_common::Durability::Replicated, + iggy_common::Durability::Persisted, + ] { + restart_keeps_acknowledged_send(durability, &[0, 1, 2]); + } + } + + /// Every replica crashes and one comes back without the partition's log, + /// so it may not elect. Under `replicated` the other two need every replica + /// to report, which the barred one does with an empty vote; it then joins + /// the view they start and catches up from it. + #[test] + fn given_every_replica_restarts_with_one_partition_log_lost_should_settle_and_accept_sends() { + for durability in [ + iggy_common::Durability::Replicated, + iggy_common::Durability::Persisted, + ] { + restart_keeps_acknowledged_send_losing(durability, &[0, 1, 2], Some(2)); + } + } + fn successful_send_reply_count(replies: &[Message<ReplyHeader>]) -> usize { replies .iter() @@ -9134,3 +9351,256 @@ mod probe_answer_divergence_tests { ); } } + +#[cfg(test)] +mod applied_divergence_resync_tests { + //! A view that contradicts an op this replica already applied fences that + //! one partition for a resync by state transfer. The WAL's parent check + //! would otherwise refuse the view's next prepare and stop the node. + + use super::*; + use bytes::Bytes; + use consensus::Status; + use iggy_binary_protocol::{ConsensusHeader, DoViewChangeHeader, StartViewHeader}; + use server_common::MessageBag; + + const CLIENT_ID: u128 = 1; + /// Backup of view 0 and primary of view 4. + const DIVERGED: u8 = 1; + /// One single-message send per op, so op `n` holds offset `n - 1`. + const SENDS: u64 = 5; + const SETTLE_STEPS: usize = 2_000; + + /// Every replica applied `SENDS` sends. + fn applied_sends(namespace: IggyNamespace) -> Simulator { + let (mut sim, client) = partition_repair_driver_tests::cluster(0x5EED_4476); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + for index in 0..SENDS { + let reply = tests::submit_and_wait_for_reply( + &mut sim, + CLIENT_ID, + 0, + client.send_messages(namespace, &[Bytes::from(format!("applied-{index}"))]), + ); + assert_eq!(reply.header().status, 0); + } + let applied = |sim: &Simulator| { + sim.replicas[usize::from(DIVERGED)] + .partition_shard(namespace) + .plane + .partitions() + .get_by_ns(&namespace) + .is_some_and(|partition| partition.consensus().commit_min() == SENDS) + }; + assert!( + (0..SETTLE_STEPS).any(|_| { + sim.step(); + applied(&sim) + }), + "replica {DIVERGED} never applied op {SENDS}" + ); + sim + } + + /// A history that leaves the cluster's at `from_op` and runs to `SENDS`, + /// newest first like a view's headers. + fn forged_suffix( + sim: &Simulator, + namespace: IggyNamespace, + from_op: u64, + ) -> Vec<PrepareHeader> { + let mut parent = tests::retained_prepare(sim, 0, namespace, from_op - 1) + .header() + .checksum; + let mut forged: Vec<PrepareHeader> = (from_op..=SENDS) + .map(|op| { + let mut header = *tests::retained_prepare(sim, 0, namespace, op).header(); + header.parent = parent; + header.timestamp += 1; + header.checksum = header.identity_checksum(); + parent = header.checksum; + header + }) + .collect(); + forged.reverse(); + forged + } + + /// A view 3 `StartView` from replica 0 carrying `headers` in order. + fn start_view(namespace: IggyNamespace, headers: &[PrepareHeader]) -> Message<StartViewHeader> { + let header_size = size_of::<StartViewHeader>(); + let total_size = header_size + size_of_val(headers); + let mut start_view = Message::<StartViewHeader>::new(total_size); + start_view.as_mut_slice()[header_size..].copy_from_slice(bytemuck::cast_slice(headers)); + let body_checksum = u128::from(iggy_common::calculate_checksum( + &start_view.as_slice()[header_size..], + )); + start_view.transmute_header(|_, header: &mut StartViewHeader| { + header.command = Command::StartView; + header.cluster = 1; + header.replica = 0; + header.view = 3; + header.group = namespace.inner(); + header.op = SENDS; + header.commit = SENDS; + header.size = u32::try_from(total_size).unwrap(); + header.checksum_body = body_checksum; + header.seal(); + }) + } + + /// The live offset frontier of the diverged replica's partition. + fn offset_frontier(sim: &Simulator, namespace: IggyNamespace) -> u64 { + sim.replicas[usize::from(DIVERGED)] + .partition_shard(namespace) + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the partition is live") + .offset_frontier() + } + + /// The partition is fenced for rebuild at `frontier`, the divergence is + /// counted, and the next sweep reports no node fault. + fn assert_resynced(sim: &Simulator, namespace: IggyNamespace, frontier: u64) { + let shard = sim.replicas[usize::from(DIVERGED)].partition_shard(namespace); + let partitions = shard.plane.partitions(); + assert!(partitions.is_tombstoned(&namespace)); + assert_eq!( + partitions + .get_io_owner(&namespace) + .and_then(|partition| partition.quarantine_frontier()), + Some(frontier), + "the rebuild must record a frontier no history past the agreed ops minted" + ); + assert_eq!(shard.metrics().partition_applied_divergence_value(), 1); + let mut namespace_scratch = Vec::new(); + assert!( + futures::executor::block_on(shard.tick_partitions(&mut namespace_scratch)).is_none(), + "a diverged partition must not stop the node" + ); + } + + #[test] + fn given_start_view_contradicting_an_applied_op_when_adopted_should_fence_the_partition_for_resync() + { + let namespace = IggyNamespace::new(1, 1, 0); + let sim = applied_sends(namespace); + let live_frontier = offset_frontier(&sim, namespace); + let forged = forged_suffix(&sim, namespace, SENDS); + let shard = sim.replicas[usize::from(DIVERGED)].partition_shard(namespace); + + futures::executor::block_on( + shard.on_message(MessageBag::StartView(start_view(namespace, &forged))), + ); + + assert_resynced(&sim, namespace, live_frontier - 1); + } + + /// Adopts a view whose history leaves the cluster's at `diverged_op` and + /// carries `window` of its newest headers plus the cluster's own `agreed` + /// ones, then checks the rebuild recorded `frontier`. + fn adopt_multi_op_divergence(diverged_op: u64, window: usize, agreed: &[u64], frontier: u64) { + let namespace = IggyNamespace::new(1, 1, 0); + let sim = applied_sends(namespace); + let mut headers: Vec<PrepareHeader> = forged_suffix(&sim, namespace, diverged_op) + .into_iter() + .take(window) + .collect(); + headers.extend( + agreed + .iter() + .map(|&op| *tests::retained_prepare(&sim, 0, namespace, op).header()), + ); + let shard = sim.replicas[usize::from(DIVERGED)].partition_shard(namespace); + + futures::executor::block_on( + shard.on_message(MessageBag::StartView(start_view(namespace, &headers))), + ); + + assert_resynced(&sim, namespace, frontier); + } + + #[test] + fn given_multi_op_divergence_with_agreed_header_when_adopted_should_record_frontier_below_first_diverged_op() + { + // Ops 4 and 5 diverge and the view still carries the cluster's op 3, so + // the offsets of ops 1..=3 are the only ones both histories minted. + adopt_multi_op_divergence(4, 2, &[3], 3); + } + + #[test] + fn given_multi_op_divergence_agreed_only_by_parent_when_adopted_should_record_frontier_below_first_diverged_op() + { + // The view's op 4 names the cluster's op 3 as parent, which is the only + // witness that op 3 is shared. + adopt_multi_op_divergence(4, 2, &[], 3); + } + + #[test] + fn given_divergence_below_the_view_window_when_adopted_should_record_no_frontier() { + // The histories split at op 2 but the view carries only ops 4 and 5, so + // nothing it holds confirms a shared op and the rebuild accepts any offer. + adopt_multi_op_divergence(2, 2, &[], 0); + } + + #[test] + fn given_merged_view_contradicting_an_applied_op_when_primary_elect_advances_should_fence_the_partition_for_resync() + { + const VIEW: u32 = 4; + let namespace = IggyNamespace::new(1, 1, 0); + let sim = applied_sends(namespace); + let live_frontier = offset_frontier(&sim, namespace); + let forged = forged_suffix(&sim, namespace, SENDS)[0]; + let shard = sim.replicas[usize::from(DIVERGED)].partition_shard(namespace); + assert_eq!( + shard + .plane + .partitions() + .get_by_ns(&namespace) + .unwrap() + .consensus() + .primary_index(VIEW), + DIVERGED + ); + let header_size = size_of::<DoViewChangeHeader>(); + let total_size = header_size + size_of::<PrepareHeader>(); + let mut do_view_change = Message::<DoViewChangeHeader>::new(total_size); + do_view_change.as_mut_slice()[header_size..].copy_from_slice(bytemuck::bytes_of(&forged)); + let body_checksum = u128::from(iggy_common::calculate_checksum( + &do_view_change.as_slice()[header_size..], + )); + let do_view_change = + do_view_change.transmute_header(|_, header: &mut DoViewChangeHeader| { + header.command = Command::DoViewChange; + header.cluster = 1; + header.replica = 2; + header.view = VIEW; + header.log_view = VIEW - 1; + header.group = namespace.inner(); + header.op = SENDS; + header.commit = SENDS; + header.present_bitset = 1; + header.size = u32::try_from(total_size).unwrap(); + header.checksum_body = body_checksum; + header.seal(); + }); + + futures::executor::block_on(shard.on_message(MessageBag::DoViewChange(do_view_change))); + let partition = shard.plane.partitions().get_by_ns(&namespace); + assert!( + partition.is_some_and(|partition| { + partition.consensus().status() == Status::ViewChange + && partition.consensus().view_log_is_pending() + }), + "the merge must park the view on the forged log" + ); + let mut namespace_scratch = Vec::new(); + assert!( + futures::executor::block_on(shard.tick_partitions(&mut namespace_scratch)).is_none() + ); + + assert_resynced(&sim, namespace, live_frontier - 1); + } +} diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs index eea28a3cf..d3474052b 100644 --- a/core/simulator/src/replica.rs +++ b/core/simulator/src/replica.rs @@ -155,7 +155,11 @@ pub fn new_shard( recovered_state: Option<VsrState>, incarnation: u128, data_dir: Option<std::path::PathBuf>, - seed_namespaces: &[(server_common::sharding::IggyNamespace, u32)], + seed_namespaces: &[( + server_common::sharding::IggyNamespace, + u32, + iggy_common::Durability, + )], applied_frontier: Arc<AppliedFrontier>, ) -> (Rc<Replica>, Option<SimMetadataBundle>) { // Metadata is single-writer, mirroring the server bootstrap. Shard 0 owns @@ -331,8 +335,13 @@ pub fn new_shard( // `materialise_partition`'s own seed is then a no-op. if shard_idx == 0 { let streams = metadata.mux_stm.streams(); - for &(namespace, created_view) in seed_namespaces { - streams.seed_namespace(namespace, namespace.inner(), created_view); + for &(namespace, created_view, durability) in seed_namespaces { + streams.seed_namespace_with_durability( + namespace, + namespace.inner(), + created_view, + durability, + ); } }
