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 b5763bc80e424a3e7b4238636eb6251d34ff5f84 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Mon Oct 12 19:02:44 2026 +0200 temp v2 --- README.md | 14 +- core/binary_protocol/src/consensus/header.rs | 24 +- core/consensus/src/dvc_merge.rs | 83 ++++- core/consensus/src/impls.rs | 484 ++++++++++++++++++++++++--- core/consensus/src/observability.rs | 20 +- core/consensus/src/view_change_quorum.rs | 26 +- core/partitions/src/iggy_partition.rs | 11 +- core/shard/src/lib.rs | 42 ++- core/shard/src/metrics.rs | 5 +- core/simulator/src/lib.rs | 155 ++++++++- core/simulator/src/replica.rs | 11 +- 11 files changed, 779 insertions(+), 96 deletions(-) diff --git a/README.md b/README.md index b7e276e2b..37f0f5492 100644 --- a/README.md +++ b/README.md @@ -284,15 +284,15 @@ 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. +What each policy survives in a cluster of `n` 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. +- `persisted` keeps every acknowledged write through any pattern of crashes and restarts. It is needed to survive `q` or more replicas crashing at once, power loss of the whole cluster, or any crash in a one-replica cluster. +- `replicated` keeps every acknowledged write as long as fewer than `q` replicas lose unflushed writes at the same time, where `q`, the replicas that acknowledge each write, is half the replicas rounded up but at most 3 (2 in a two-replica cluster). A replica that returns with that partition awaiting a state transfer counts as one of them: its log is not used, even if its WAL still holds ops. +- Graceful restarts flush the WAL first and lose nothing, one at a time or all together. A crash (SIGKILL, OOM kill) loses the WAL writes not yet handed to the OS, and power loss also loses what the OS had not written to disk. Under `replicated`, `q` or more such crashes at once can lose the most recently acknowledged writes. - 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 no primary answers, the restarted replicas hold an election and mark their logs as possibly incomplete. A new view starts only from `n - q + 1` replicas whose logs are not marked (2 in a two-replica cluster, a majority up to six replicas), or once every replica has reported, and the partition stays unavailable until then. + After a full-cluster restart that means every replica must be back. A replica whose partition awaits a state transfer still reports without a log while it is recovering; once it has joined a view it stays out of later elections until its transfer completes. A replica that never returns halts the partition indefinitely (no operator override yet). + Do not run mixed versions: an older replica can read a newer replica's empty vote as a log without the latest writes and drop acknowledged writes. Upgrade every replica together, with the cluster down. - 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. diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs index a60a485c8..46fc98015 100644 --- a/core/binary_protocol/src/consensus/header.rs +++ b/core/binary_protocol/src/consensus/header.rs @@ -1530,7 +1530,7 @@ pub struct DoViewChangeHeader { pub group: u64, /// View when status was last normal (key for log selection). pub log_view: u32, - /// Bit 0, [`Self::FLAG_LOG_UNTRUSTED`]: the sender recovered a `replicated` + /// Bit 0, [`Self::FLAG_UNTRUSTED_LOG`]: 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. @@ -1538,6 +1538,10 @@ pub struct DoViewChangeHeader { /// 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. + /// + /// Every bit is must-understand: `validate` rejects one outside + /// [`Self::FLAGS_KNOWN`]. `reserved` is not checked, so a later field carved + /// from it must be one an older receiver can safely ignore. pub flags: u8, pub reserved: [u8; 67], /// Bit `i` set means the sender proves it never prepared suffix entry `i`, so @@ -1663,14 +1667,14 @@ impl ConsensusHeader for DoViewChangeHeader { impl DoViewChangeHeader { /// See [`Self::flags`]. - pub const FLAG_LOG_UNTRUSTED: u8 = 1; + pub const FLAG_UNTRUSTED_LOG: u8 = 1; /// Every flag bit this build understands. - pub const FLAGS_KNOWN: u8 = Self::FLAG_LOG_UNTRUSTED; + pub const FLAGS_KNOWN: u8 = Self::FLAG_UNTRUSTED_LOG; /// 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 + pub const fn untrusted_log(&self) -> bool { + self.flags & Self::FLAG_UNTRUSTED_LOG != 0 } /// Number of `PrepareHeader`s in the body. @@ -3093,11 +3097,11 @@ mod tests { #[test] fn given_untrusted_flag_when_round_tripped_should_keep_flag_and_field_offsets() { - let header = do_view_change(DoViewChangeHeader::FLAG_LOG_UNTRUSTED); + let header = do_view_change(DoViewChangeHeader::FLAG_UNTRUSTED_LOG); 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[156], DoViewChangeHeader::FLAG_UNTRUSTED_LOG); 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()); @@ -3107,7 +3111,7 @@ mod tests { 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!(decoded.untrusted_log()); assert_eq!(decoded.verify_frame(), Ok(())); assert_eq!(decoded.validate(), Ok(())); } @@ -3116,14 +3120,14 @@ mod tests { 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!(!header.untrusted_log()); 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)); + let header = do_view_change(DoViewChangeHeader::FLAG_UNTRUSTED_LOG | (1 << bit)); assert!( matches!(header.validate(), Err(ConsensusError::InvalidField(_))), "flag bit {bit} has no meaning yet and must not be read past" diff --git a/core/consensus/src/dvc_merge.rs b/core/consensus/src/dvc_merge.rs index 81e69747b..388d4f977 100644 --- a/core/consensus/src/dvc_merge.rs +++ b/core/consensus/src/dvc_merge.rs @@ -407,7 +407,7 @@ fn committed_elsewhere( } /// 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. +/// ops they acknowledged, as a sender with [`StoredDvc::untrusted_log`] 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 @@ -422,15 +422,16 @@ fn committed_elsewhere( /// 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. +/// body would settle, so once every replica has reported, at least one of them +/// untrusted, 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(); + let trusted = dvc_iter(quorum).filter(|dvc| !dvc.untrusted_log).count(); if trusted >= quorums.view_change { let trusted_quorums = MergeQuorums { absence_nacks: true, @@ -440,12 +441,17 @@ pub fn merge_view_change(quorum: &DvcQuorumArray, quorums: MergeQuorums) -> Merg 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) { + for dvc in dvc_iter(quorum).filter(|dvc| !dvc.untrusted_log) { dvc_record(&mut trusted_only, dvc.clone()); } merge_dvc_quorum(&trusted_only, trusted_quorums) }; - if matches!(outcome, MergeOutcome::Ready(_)) || received < quorums.replica_count { + // With every sender trusted the all-n merge differs only by dropping + // absence nacks, which would turn a detected deadlock into a guess. + if matches!(outcome, MergeOutcome::Ready(_)) + || received < quorums.replica_count + || trusted == received + { return outcome; } } else if received < quorums.replica_count { @@ -664,7 +670,7 @@ mod tests { op, commit, suffix, - log_untrusted: false, + untrusted_log: false, } } @@ -1132,7 +1138,7 @@ mod tests { fn untrusted(dvc: StoredDvc) -> StoredDvc { StoredDvc { - log_untrusted: true, + untrusted_log: true, ..dvc } } @@ -1255,6 +1261,67 @@ mod tests { assert_eq!(merged.op_head, 9); } + #[test] + fn given_every_sender_trusted_when_the_merge_deadlocks_should_stay_deadlocked() { + // Two absent heads nack op 6, which replica 0 proves committed: a lie or + // a wrong bitset. Retrying without absence nacks would adopt the op and + // hide the contradiction, and no untrusted sender justifies that retry. + let mut quorum = dvc_quorum_array_empty(); + dvc_record( + &mut quorum, + dvc(0, 1, 6, 6, suffix_all_present(suffix_headers(6, 6, 1))), + ); + dvc_record( + &mut quorum, + dvc(1, 1, 4, 4, suffix_all_present(suffix_headers(4, 4, 1))), + ); + dvc_record( + &mut quorum, + dvc(2, 1, 4, 4, suffix_all_present(suffix_headers(4, 4, 1))), + ); + let without_absence_nacks = MergeQuorums { + view_change: 3, + absence_nacks: false, + ..quorums_r3() + }; + assert!( + matches!( + merge_dvc_quorum(&quorum, without_absence_nacks), + MergeOutcome::Ready(_) + ), + "the all-replica rule must decide differently for this case to pin anything" + ); + + assert_eq!( + merge_view_change(&quorum, quorums_r3()), + MergeOutcome::Deadlocked { undecided_op: 6 } + ); + } + + #[test] + fn given_a_seated_empty_vote_when_the_same_replica_sends_its_log_should_replace_it() { + let mut quorum = dvc_quorum_array_empty(); + assert!(dvc_record(&mut quorum, empty_vote(2))); + assert!(!dvc_record(&mut quorum, empty_vote(2))); + + let real = untrusted(dvc(2, 1, 9, 4, suffix_all_present(suffix_headers(4, 9, 1)))); + assert!( + dvc_record(&mut quorum, real), + "the replica's log was hidden behind its empty vote" + ); + let seated = quorum[2].as_ref().expect("replica 2 reported"); + assert_eq!(seated.op, 9); + + assert!(!dvc_record(&mut quorum, empty_vote(2))); + assert!(!dvc_record( + &mut quorum, + dvc(2, 1, 5, 4, suffix_all_present(suffix_headers(4, 5, 1))) + )); + let seated = quorum[2].as_ref().expect("replica 2 reported"); + assert_eq!(seated.op, 9, "a seated log was replaced"); + assert_eq!(dvc_count(&quorum), 1); + } + #[test] fn given_an_empty_vote_when_every_replica_reports_should_count_it_and_keep_every_op() { let mut quorum = dvc_quorum_array_empty(); diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index b1a41bd9c..1e7e62cdd 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -1075,8 +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, + /// Stamped as `DoViewChangeHeader::FLAG_UNTRUSTED_LOG`. + untrusted_log: bool, }, /// Broadcast a `RequestStartView` probe (recovering replica asking for /// the current view's `StartView`; only that view's primary answers). @@ -1162,7 +1162,7 @@ impl VsrAction { log_view: 0, op: 0, commit: 0, - log_untrusted: true, + untrusted_log: true, suffix, .. } if suffix.is_empty() @@ -1301,12 +1301,28 @@ where /// with a log shorter than what it acknowledged. See /// [`Self::set_recovered_log_complete`]. recovered_log_complete: Cell<bool>, + /// This incarnation booted on the log it read back from disk and has not + /// been `Normal` since. Set by [`Self::join`] with + /// [`JoinMode::ProbeAsBackup`], cleared on the first `Normal`. + /// + /// A view probe begun later, by a replica that never restarted, finds it + /// clear: that log lost nothing, so an unanswered probe is no reason to + /// distrust it, and doing so could halt a group one crash short of its + /// fault budget. + /// + /// Adopting a `StartView` and starting a view as primary are the clears + /// that matter. Those in [`Self::init`] and [`Self::init_as_backup`] are + /// only a safeguard: each runs at boot, before anything but `join` sets + /// the flag, and `join` sets it after `init_as_backup` returns. + rejoined_on_recovered_log: Cell<bool>, /// This replica left `Recovering` through the probe-exhaustion election - /// while [`Self::recovered_log_complete`] was false: its log may lack ops it + /// on a log read back after a restart (see + /// [`Self::rejoined_on_recovered_log`]) 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 + /// toward a trusted 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. @@ -1632,6 +1648,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { await_state_transfer, } => { self.init_as_backup(); + self.rejoined_on_recovered_log.set(true); self.begin_view_probe(); if await_state_transfer { self.begin_state_transfer_await(); @@ -1706,6 +1723,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { probe_attempts_max: Cell::new(PROBE_ATTEMPTS_MAX), recovery_election_allowed: Cell::new(true), recovered_log_complete: Cell::new(true), + rejoined_on_recovered_log: Cell::new(false), untrusted_log: Cell::new(false), untrusted_view_change_timeouts: Cell::new(0), local_dvc_suffix: RefCell::new(None), @@ -1798,11 +1816,11 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { /// 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`]. + /// With `false`, a restarted replica whose boot 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); } @@ -1821,6 +1839,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { pub fn init(&self) { self.status.set(Status::Normal); + self.rejoined_on_recovered_log.set(false); let mut timeouts = self.timeouts.borrow_mut(); if self.is_primary() { // Prepare is deliberately NOT armed here: a fresh primary has an @@ -1849,6 +1868,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { /// it stays quorum-invisible until a `StartView` answers. pub fn init_as_backup(&self) { self.status.set(Status::Normal); + self.rejoined_on_recovered_log.set(false); self.ceded_primaryship.set(true); self.timeouts .borrow_mut() @@ -2782,8 +2802,11 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // 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() { + // committed ops that only it still has. Only a log read + // back after a restart can be short of what it acked. + if self.rejoined_on_recovered_log.get() + && !self.recovered_log_complete.get() + { tracing::info!( replica = self.replica, namespace_raw = self.group, @@ -3002,7 +3025,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // NOT the snapshot the first send used: `build_do_view_change` re-reads // `local_dvc_suffix()`, whose tag can have moved since, in which case it // answers EMPTY, retracting every nack and body offer already sent. - // Survivable only because `dvc_record` drops a duplicate sender, so the + // Survivable only because `dvc_record` drops a duplicate real vote, so the // candidate keeps the first vote. Allow a retransmit to replace a seated vote // and this must pin the snapshot instead. let action = self.build_do_view_change(self.primary_index(self.view.get())); @@ -3282,7 +3305,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(), + untrusted_log: self.untrusted_log.get(), }; dvc_record( &mut self.do_view_change_from_all_replicas.borrow_mut(), @@ -3331,12 +3354,13 @@ 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(), + untrusted_log: self.untrusted_log.get(), } } - /// The `DoViewChange` a recovering replica barred from electing sends for - /// `view`, or `None` when it must stay silent. + /// The `DoViewChange` a recovering replica barred from electing sends in + /// answer to a `StartViewChange` for `view`, or `None` when it must stay + /// silent. Never in answer to a `DoViewChange`: see `handle_do_view_change`. /// /// Without it a group whose every log is untrusted could never start a /// view, since that takes every replica reporting and a barred replica @@ -3365,7 +3389,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { commit: 0, group: self.group, suffix: DvcSuffix::empty(), - log_untrusted: true, + untrusted_log: true, }; emit_sim_event( SimEventKind::ControlMessageScheduled, @@ -3420,12 +3444,12 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { ) -> Vec<VsrAction> { assert_eq!(header.group, self.group, "DVC routed to wrong group"); // Quorum-invisible while recovering (see handle_start_view_change): - // a recovering replica must not collect DVCs and crown itself. + // a recovering replica must not collect DVCs and crown itself. Nor does + // it answer one with its empty vote: DVCs are broadcast, so two barred + // replicas would answer each other's empty votes forever. The + // candidate's SVC retransmits keep soliciting the vote instead. if self.status.get() == Status::Recovering { - return self - .empty_do_view_change(plane, header.view) - .into_iter() - .collect(); + return Vec::new(); } let from_replica = header.replica; let msg_view = header.view; @@ -3476,7 +3500,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(), + untrusted_log: self.untrusted_log.get(), }; dvc_record( &mut self.do_view_change_from_all_replicas.borrow_mut(), @@ -3491,9 +3515,12 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { op: msg_op, commit: msg_commit, suffix: msg_suffix, - log_untrusted: header.log_untrusted(), + untrusted_log: header.untrusted_log(), }; - dvc_record(&mut self.do_view_change_from_all_replicas.borrow_mut(), dvc); + // A retransmit changes nothing the last merge did not already see. + if !dvc_record(&mut self.do_view_change_from_all_replicas.borrow_mut(), dvc) { + return actions; + } // `complete_view_change_as_primary` latches only once the merge decides, // so an undecidable quorum re-merges as each further DVC lands. @@ -3851,6 +3878,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.view.set(msg_view); self.log_view.set(msg_view); self.status.set(Status::Normal); + self.rejoined_on_recovered_log.set(false); self.ceded_primaryship.set(false); // The view's log now comes from a settled primary, and repair fills this // replica's holes from it. @@ -4235,6 +4263,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.invalidate_local_dvc_suffix(); self.status.set(Status::Normal); + self.rejoined_on_recovered_log.set(false); self.ceded_primaryship.set(false); // Merged from a trusted quorum or from every replica, so no surviving log // was left out. @@ -6461,7 +6490,7 @@ mod view_source_tests { op, commit, suffix: DvcSuffix::new(headers, 0, present), - log_untrusted: false, + untrusted_log: false, } } @@ -6878,10 +6907,15 @@ mod recovery_election_tests { Message::<H>::new(HEADER_SIZE + body).transmute_header(|_, header: &mut H| fill(header)) } + /// Messages one `Cluster::deliver` may carry before it counts as a loop: + /// orders of magnitude above what one tick of three replicas provokes. + const DELIVERIES_PER_ROUND_MAX: usize = 10_000; + struct DeliveredDoViewChange { from: u8, + to: u8, view: u32, - log_untrusted: bool, + untrusted_log: bool, } struct Cluster { @@ -6889,6 +6923,11 @@ mod recovery_election_tests { plane: PlaneKind, /// Every `DoViewChange` delivered. do_view_changes: Vec<DeliveredDoViewChange>, + /// Deliver each `DoViewChange` to every peer rather than to its target, + /// as the shard dispatches it. + broadcast_do_view_changes: bool, + /// Messages delivered so far. + delivered: usize, } impl Cluster { @@ -6903,11 +6942,19 @@ mod recovery_election_tests { mut outbox: VecDeque<(u8, VsrAction)>, reachable: impl Fn(u8, u8) -> bool, ) { + let mut round = 0; while let Some((from, action)) = outbox.pop_front() { for to in 0..REPLICA_COUNT { if to == from || !reachable(from, to) { continue; } + round += 1; + assert!( + round <= DELIVERIES_PER_ROUND_MAX, + "unbounded message loop: {round} messages provoked in one round, \ + the last {action:?} from replica {from} to {to}" + ); + self.delivered += 1; for reply in self.receive(from, to, &action) { outbox.push_back((to, reply)); } @@ -6950,8 +6997,8 @@ mod recovery_election_tests { commit, group, suffix, - log_untrusted, - } if *target == to => { + untrusted_log, + } if self.broadcast_do_view_changes || *target == to => { let mut dvc = message(suffix.encoded_len(), |header: &mut DoViewChangeHeader| { header.command = Command::DoViewChange; @@ -6964,8 +7011,8 @@ mod recovery_election_tests { header.op = *op; header.commit = *commit; header.group = *group; - if *log_untrusted { - header.flags = DoViewChangeHeader::FLAG_LOG_UNTRUSTED; + if *untrusted_log { + header.flags = DoViewChangeHeader::FLAG_UNTRUSTED_LOG; } header.nack_bitset = suffix.nack_bitset(); header.present_bitset = suffix.present_bitset(); @@ -6973,8 +7020,9 @@ mod recovery_election_tests { suffix.encode_into(&mut dvc.as_mut_slice()[HEADER_SIZE..]); self.do_view_changes.push(DeliveredDoViewChange { from, + to, view: *view, - log_untrusted: *log_untrusted, + untrusted_log: *untrusted_log, }); self.replica(to).handle_do_view_change( plane, @@ -7131,6 +7179,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; assert_eq!( cluster.replica(RESTARTED_PRIMARY).status(), @@ -7167,6 +7217,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(TICKS, |_, _| true); @@ -7188,6 +7240,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(TICKS, |from, to| from != SURVIVOR && to != SURVIVOR); @@ -7222,6 +7276,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(TICKS, |_, _| true); @@ -7248,6 +7304,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(TICKS, |_, _| true); @@ -7264,7 +7322,7 @@ mod recovery_election_tests { .do_view_changes .iter() .filter(|delivered| delivered.from == SURVIVOR) - .all(|delivered| delivered.log_untrusted), + .all(|delivered| delivered.untrusted_log), "an empty vote must never read as a trusted log" ); } @@ -7286,6 +7344,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; for tick in 0..TICKS { @@ -7327,6 +7387,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; let view = cluster.replica(SURVIVOR).view(); @@ -7364,6 +7426,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(2 * TICKS, |_, _| true); @@ -7386,6 +7450,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(TICKS, |_, _| true); @@ -7395,7 +7461,7 @@ mod recovery_election_tests { cluster .do_view_changes .iter() - .all(|delivered| delivered.from != SURVIVOR && !delivered.log_untrusted) + .all(|delivered| delivered.from != SURVIVOR && !delivered.untrusted_log) ); } @@ -7413,6 +7479,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(TICKS, |from, to| from != SURVIVOR && to != SURVIVOR); @@ -7454,9 +7522,11 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; let do_view_change = - |head: u64, commit: u64, log_untrusted: bool| VsrAction::SendDoViewChange { + |head: u64, commit: u64, untrusted_log: bool| VsrAction::SendDoViewChange { view, target: LAGGING_BACKUP, log_view: OLD_VIEW, @@ -7464,7 +7534,7 @@ mod recovery_election_tests { commit, group: group(), suffix: suffix_all_present((commit..=head).rev().map(prepare).collect()), - log_untrusted, + untrusted_log, }; cluster.receive( @@ -7534,6 +7604,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; cluster.run(TICKS, |from, to| from != SURVIVOR && to != SURVIVOR); @@ -7542,7 +7614,7 @@ mod recovery_election_tests { assert!(cluster.replica(RESTARTED_PRIMARY).has_untrusted_log()); for delivered in &cluster.do_view_changes { assert_eq!( - delivered.log_untrusted, + delivered.untrusted_log, delivered.from == RESTARTED_PRIMARY, "replica {} sent a DoViewChange misreporting its log", delivered.from @@ -7597,6 +7669,8 @@ mod recovery_election_tests { ], plane, do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, }; let svc = VsrAction::SendStartViewChange { view, @@ -7604,7 +7678,7 @@ mod recovery_election_tests { }; cluster.receive(SURVIVOR, LAGGING_BACKUP, &svc); let do_view_change = - |head: u64, commit: u64, log_untrusted: bool| VsrAction::SendDoViewChange { + |head: u64, commit: u64, untrusted_log: bool| VsrAction::SendDoViewChange { view, target: LAGGING_BACKUP, log_view: OLD_VIEW, @@ -7612,7 +7686,7 @@ mod recovery_election_tests { commit, group: group(), suffix: suffix_all_present((commit..=head).rev().map(prepare).collect()), - log_untrusted, + untrusted_log, }; cluster.receive( @@ -7638,4 +7712,334 @@ mod recovery_election_tests { "the merge took the untrusted sender's longer log" ); } + + fn barred(replica: u8, head: u64) -> VsrConsensus<NoopBus, LocalPipeline> { + let consensus = restarted(replica, head, false); + consensus.set_recovery_election_allowed(false); + consensus + } + + /// Two replicas come back barred from electing. The shard broadcasts every + /// `DoViewChange`, so each hears the other's empty vote, and answering one + /// with its own would bounce empty votes between them forever. While the + /// third replica's view change cannot complete they stay quiet and + /// `Recovering`; once it can, the group settles. + #[test] + fn given_two_barred_replicas_when_do_view_changes_are_broadcast_should_not_answer_each_other() { + let plane = PlaneKind::Partitions; + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + barred(LAGGING_BACKUP, LAGGING_HEAD), + barred(SURVIVOR, RECOVERED_HEAD), + ], + plane, + do_view_changes: Vec::new(), + broadcast_do_view_changes: true, + delivered: 0, + }; + + // The electing replica's view-change traffic reaches the barred pair, + // their votes reach only each other. + cluster.run(TICKS, |from, to| { + from == RESTARTED_PRIMARY || to != RESTARTED_PRIMARY + }); + + assert!( + cluster.do_view_changes.iter().any(|delivered| { + delivered.from != RESTARTED_PRIMARY && delivered.to != RESTARTED_PRIMARY + }), + "no empty vote reached the other barred replica" + ); + for index in [LAGGING_BACKUP, SURVIVOR] { + assert_eq!(cluster.replica(index).status(), Status::Recovering); + } + let quiet = cluster.delivered; + assert!( + quiet <= usize::try_from(TICKS).unwrap() * 8, + "{quiet} messages in {TICKS} ticks" + ); + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + } + + /// Lagging backups that never restarted, then heard a newer view and + /// probed for it. With the third replica gone nobody answers, and the + /// fallback election must treat their logs as the intact ones they are: + /// marking them untrusted would halt a group that lost only one replica. + #[test] + fn given_replicas_that_never_restarted_when_a_runtime_probe_exhausts_should_elect_trusted_without_the_third() + { + let plane = PlaneKind::Partitions; + let probing = |replica: u8| { + let mut consensus = VsrConsensus::new( + 1, + replica, + REPLICA_COUNT, + group(), + NoopBus, + LocalPipeline::new(), + ); + consensus.set_view(OLD_VIEW); + consensus.set_log_view(OLD_VIEW); + consensus.sequencer().set_sequence(COMMITTED); + consensus.restore_commit_state(COMMITTED - 1, COMMITTED - 1); + consensus.set_last_prepare_checksum(prepare(COMMITTED).checksum); + consensus.set_local_dvc_suffix(suffix_all_present( + (COMMITTED - 1..=COMMITTED).rev().map(prepare).collect(), + )); + consensus.set_recovered_log_complete(false); + consensus.init(); + consensus.begin_view_probe(); + consensus + }; + let mut cluster = Cluster { + replicas: vec![ + probing(RESTARTED_PRIMARY), + probing(LAGGING_BACKUP), + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, + }; + + 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::Normal, "replica {index} halted"); + assert!(replica.log_view() > OLD_VIEW); + assert!(!replica.has_untrusted_log()); + } + assert!( + cluster + .do_view_changes + .iter() + .all(|delivered| !delivered.untrusted_log) + ); + } + + /// The two continuing replicas of a settled view probe at runtime with the + /// third gone. Both restarted earlier on logs that may lack acknowledged + /// ops, but a settled view vouched for them since, so the election must + /// treat their logs as trusted rather than halt. + fn assert_runtime_probe_elects_trusted_without_survivor(cluster: &mut Cluster) { + let settled_view = cluster.replica(RESTARTED_PRIMARY).view(); + for index in [RESTARTED_PRIMARY, LAGGING_BACKUP] { + cluster.replica(index).begin_view_probe(); + } + let elected_from = cluster.do_view_changes.len(); + + 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::Normal, "replica {index} halted"); + assert!(replica.log_view() > settled_view); + assert!(!replica.has_untrusted_log()); + } + assert!( + cluster.do_view_changes[elected_from..] + .iter() + .all(|delivered| !delivered.untrusted_log), + "a replica that settled after its restart still marked its log untrusted" + ); + } + + /// Restarted replicas whose probe a live primary answers adopt its + /// `StartView`, which vouches for their logs. A later runtime probe that + /// exhausts with that primary gone must elect them trusted. + #[test] + fn given_restarted_replicas_that_adopted_a_start_view_when_a_runtime_probe_exhausts_should_elect_trusted_without_the_third() + { + const SETTLED_VIEW: u32 = OLD_VIEW + 2; + let plane = PlaneKind::Partitions; + let mut primary = VsrConsensus::new( + 1, + SURVIVOR, + REPLICA_COUNT, + group(), + NoopBus, + LocalPipeline::new(), + ); + primary.set_view(SETTLED_VIEW); + primary.set_log_view(SETTLED_VIEW); + primary.sequencer().set_sequence(COMMITTED); + primary.restore_commit_state(COMMITTED - 1, COMMITTED - 1); + primary.set_last_prepare_checksum(prepare(COMMITTED).checksum); + primary.set_local_dvc_suffix(suffix_all_present( + (COMMITTED - 1..=COMMITTED).rev().map(prepare).collect(), + )); + primary.init(); + assert!(primary.is_primary()); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + restarted(LAGGING_BACKUP, COMMITTED, false), + primary, + ], + plane, + do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, + }; + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + assert_eq!(cluster.replica(SURVIVOR).view(), SETTLED_VIEW); + assert!( + cluster.do_view_changes.is_empty(), + "the probe was not answered" + ); + assert_runtime_probe_elects_trusted_without_survivor(&mut cluster); + } + + /// A full restart settles on untrusted logs once every replica reports: one + /// continuing replica starts the view as its primary, the other adopts its + /// `StartView`. A later runtime probe that exhausts with the third gone must + /// elect both trusted. + #[test] + fn given_full_restart_that_settled_when_a_runtime_probe_exhausts_should_elect_trusted_without_the_third() + { + let plane = PlaneKind::Partitions; + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, COMMITTED, false), + restarted(LAGGING_BACKUP, COMMITTED, false), + restarted(SURVIVOR, COMMITTED, false), + ], + plane, + do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, + }; + + cluster.run(TICKS, |_, _| true); + + cluster.assert_settled(COMMITTED); + assert!( + cluster + .do_view_changes + .iter() + .any(|delivered| delivered.untrusted_log), + "the full restart elected without marking its logs" + ); + assert!( + [RESTARTED_PRIMARY, LAGGING_BACKUP] + .iter() + .any(|index| cluster.replica(*index).is_primary()), + "the settled primary is the replica the runtime probe leaves out" + ); + assert_runtime_probe_elects_trusted_without_survivor(&mut cluster); + } + + /// An untrusted candidate can learn its view from a peer's `DoViewChange` + /// before enough `StartViewChange`s arrive to send its own. The vote it + /// records for itself must still say untrusted, so one trusted peer is not + /// a trusted quorum and it waits for every replica. + #[test] + fn given_untrusted_candidate_when_a_peer_do_view_change_brings_it_into_the_view_should_record_its_own_vote_untrusted() + { + let plane = PlaneKind::Partitions; + let candidate = restarted(LAGGING_BACKUP, LAGGING_HEAD, false); + while candidate.status() == Status::Recovering { + candidate.tick(plane); + } + assert!(candidate.has_untrusted_log()); + let view = (candidate.view() + 1..=candidate.view() + u32::from(REPLICA_COUNT)) + .find(|view| candidate.primary_index(*view) == LAGGING_BACKUP) + .expect("every replica leads one of any run of replica-count views"); + let mut cluster = Cluster { + replicas: vec![ + restarted(RESTARTED_PRIMARY, RECOVERED_HEAD, false), + candidate, + survivor_in_view_change(plane), + ], + plane, + do_view_changes: Vec::new(), + broadcast_do_view_changes: false, + delivered: 0, + }; + let do_view_change = + |head: u64, commit: u64, untrusted_log: 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()), + untrusted_log, + }; + + cluster.receive( + SURVIVOR, + LAGGING_BACKUP, + &do_view_change(COMMITTED, COMMITTED - 1, false), + ); + let candidate = cluster.replica(LAGGING_BACKUP); + assert_eq!(candidate.view(), view); + assert!( + candidate.do_view_change_from_all_replicas.borrow()[usize::from(LAGGING_BACKUP)] + .as_ref() + .expect("the candidate recorded its own vote") + .untrusted_log + ); + assert!( + !candidate.view_log_is_pending(), + "one trusted peer completed a quorum with the candidate's untrusted log" + ); + + cluster.receive( + RESTARTED_PRIMARY, + LAGGING_BACKUP, + &do_view_change(RECOVERED_HEAD, RECOVERED_HEAD - 1, true), + ); + assert_eq!( + cluster + .replica(LAGGING_BACKUP) + .with_pending_view_log(|log| log.op_head), + Some(COMMITTED) + ); + } + + /// A barred replica votes only for a view it has not moved past and does + /// not lead. + #[test] + fn given_barred_replica_when_a_view_change_is_older_or_its_own_should_send_no_empty_vote() { + let plane = PlaneKind::Partitions; + let barred = barred(SURVIVOR, COMMITTED); + let older = OLD_VIEW - 2; + let led = OLD_VIEW + 2; + assert_ne!(barred.primary_index(older), SURVIVOR); + assert_eq!(barred.primary_index(led), SURVIVOR); + for (view, votes) in [ + (older, false), + (OLD_VIEW, true), + (OLD_VIEW + 1, true), + (led, false), + ] { + let svc = message(0, |header: &mut StartViewChangeHeader| { + header.command = Command::StartViewChange; + header.cluster = 1; + header.size = u32::try_from(HEADER_SIZE).unwrap(); + header.replica = RESTARTED_PRIMARY; + header.view = view; + header.group = group(); + }); + let actions = barred.handle_start_view_change(plane, svc.header()); + assert_eq!( + actions.iter().any(VsrAction::is_empty_vote), + votes, + "view {view}: {actions:?}" + ); + assert_eq!(barred.status(), Status::Recovering); + assert_eq!(barred.view(), OLD_VIEW); + } + } } diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs index 44141b3fb..4720fdbd7 100644 --- a/core/consensus/src/observability.rs +++ b/core/consensus/src/observability.rs @@ -318,6 +318,10 @@ pub struct ControlActionLogEvent { pub target_replica: Option<u8>, pub op: Option<u64>, pub commit: Option<u64>, + /// The flag a `DoViewChange` carries, which for an empty vote differs from + /// [`ReplicaLogContext::untrusted_log`]: a barred replica never elected, yet + /// its vote is untrusted. + pub untrusted_log: Option<bool>, } impl ControlActionLogEvent { @@ -330,6 +334,7 @@ impl ControlActionLogEvent { target_replica: None, op: None, commit: None, + untrusted_log: None, }, VsrAction::SendStartViewChange { .. } => Self { replica, @@ -337,15 +342,21 @@ impl ControlActionLogEvent { target_replica: None, op: None, commit: None, + untrusted_log: None, }, VsrAction::SendDoViewChange { - target, op, commit, .. + target, + op, + commit, + untrusted_log, + .. } => Self { replica, action: ControlActionKind::SendDoViewChange, target_replica: Some(target), op: Some(op), commit: Some(commit), + untrusted_log: Some(untrusted_log), }, VsrAction::SendStartView { op, commit, .. } => Self { replica, @@ -353,6 +364,7 @@ impl ControlActionLogEvent { target_replica: None, op: Some(op), commit: Some(commit), + untrusted_log: None, }, VsrAction::SendPrepareOk { target, from_op, .. @@ -362,6 +374,7 @@ impl ControlActionLogEvent { target_replica: Some(target), op: Some(from_op), commit: None, + untrusted_log: None, }, VsrAction::RetransmitPrepares { .. } => Self { replica, @@ -369,6 +382,7 @@ impl ControlActionLogEvent { target_replica: None, op: None, commit: None, + untrusted_log: None, }, VsrAction::RebuildPipeline { from_op, to_op, .. } => Self { replica, @@ -376,6 +390,7 @@ impl ControlActionLogEvent { target_replica: None, op: Some(from_op), commit: Some(to_op), + untrusted_log: None, }, VsrAction::CommitJournal => Self { replica, @@ -383,6 +398,7 @@ impl ControlActionLogEvent { target_replica: None, op: None, commit: None, + untrusted_log: None, }, VsrAction::SendCommit { commit, .. } => Self { replica, @@ -390,6 +406,7 @@ impl ControlActionLogEvent { target_replica: None, op: None, commit: Some(commit), + untrusted_log: None, }, } } @@ -982,6 +999,7 @@ impl StructuredSimEvent for ControlActionLogEvent { target_replica = self.target_replica.unwrap_or_default(), op = self.op.unwrap_or_default(), action_commit = self.commit.unwrap_or_default(), + action_untrusted_log = self.untrusted_log.unwrap_or_default(), ); } } diff --git a/core/consensus/src/view_change_quorum.rs b/core/consensus/src/view_change_quorum.rs index 4a77f83e7..f2996734f 100644 --- a/core/consensus/src/view_change_quorum.rs +++ b/core/consensus/src/view_change_quorum.rs @@ -213,7 +213,20 @@ pub struct StoredDvc { 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, + pub untrusted_log: bool, +} + +impl StoredDvc { + /// Whether this is the log-less vote a recovering replica barred from + /// electing casts. See `VsrAction::is_empty_vote`. + #[must_use] + pub fn is_empty_vote(&self) -> bool { + self.untrusted_log + && self.log_view == 0 + && self.op == 0 + && self.commit == 0 + && self.suffix.is_empty() + } } /// Array type for storing DVC messages from all replicas. @@ -227,10 +240,17 @@ pub fn dvc_quorum_array_empty() -> DvcQuorumArray { } /// Record a DVC in the array. Returns true if this is a new entry (not duplicate). +/// +/// A real vote replaces the same sender's empty vote: the sender has since +/// recovered a log to vote with, and keeping the empty one would hide it from a +/// merge still waiting on that log. pub fn dvc_record(array: &mut DvcQuorumArray, dvc: StoredDvc) -> bool { let slot = &mut array[dvc.replica as usize]; - if slot.is_some() { - return false; // Duplicate + if slot + .as_ref() + .is_some_and(|seated| !seated.is_empty_vote() || dvc.is_empty_vote()) + { + return false; } *slot = Some(dvc); true diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 11a1020ee..3734c590a 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -2690,6 +2690,15 @@ where } } + /// Mark this partition's files lost, as boot does when it finds the + /// on-disk `materialization_missing` record, which in-memory storage has no + /// directory to hold. + #[cfg(feature = "fault-injection")] + pub fn force_requires_state_transfer_for_test(&mut self) { + self.materialization_missing = true; + self.ensure_materialization_recovery(); + } + /// 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 { @@ -12243,6 +12252,7 @@ mod tests { use iggy_binary_protocol::{ Command, CommitHeader, ReplyHeader, StartViewHeader, WireConsumer, WireEncode, WireName, }; + use iggy_common::Durability::{Persisted, Replicated}; use journal::DurableAppend; use journal::durable_storage::DiskStorage; use message_bus::{BusMessage, SendError}; @@ -14096,7 +14106,6 @@ mod tests { #[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(), diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 3e766cbfb..3d0fcd6fb 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -11732,7 +11732,7 @@ async fn dispatch_vsr_actions<B, P, J>( commit, group, suffix, - log_untrusted, + untrusted_log, } => { let header_size = size_of::<DoViewChangeHeader>(); let total_size = header_size + suffix.encoded_len(); @@ -11752,8 +11752,8 @@ 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; + if *untrusted_log { + h.flags |= DoViewChangeHeader::FLAG_UNTRUSTED_LOG; } h.nack_bitset = nack_bitset; h.present_bitset = present_bitset; @@ -13923,7 +13923,7 @@ mod untrusted_do_view_change_tests { const GROUP: u64 = 42; - fn do_view_change(op: u64, log_untrusted: bool) -> VsrAction { + fn do_view_change(op: u64, untrusted_log: bool) -> VsrAction { VsrAction::SendDoViewChange { view: 1, target: 1, @@ -13932,7 +13932,7 @@ mod untrusted_do_view_change_tests { commit: 0, group: GROUP, suffix: DvcSuffix::empty(), - log_untrusted, + untrusted_log, } } @@ -13957,12 +13957,12 @@ mod untrusted_do_view_change_tests { IggyByteSize::from(1024 * 1024), )); - for log_untrusted in [true, false] { + for untrusted_log in [true, false] { sent.borrow_mut().clear(); dispatch_partition_wire_actions::<_, _, PrepareJournal, _>( partition.consensus(), &partition, - vec![do_view_change(0, log_untrusted)], + vec![do_view_change(0, untrusted_log)], ) .await; let frames = sent.borrow(); @@ -13972,7 +13972,7 @@ mod untrusted_do_view_change_tests { frame.as_slice(), )) .expect("a dispatched DoViewChange decodes"); - assert_eq!(message.header().log_untrusted(), log_untrusted); + assert_eq!(message.header().untrusted_log(), untrusted_log); assert_eq!(message.header().verify_frame(), Ok(())); } } @@ -13991,10 +13991,28 @@ mod untrusted_do_view_change_tests { !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" - ); + let flagged = |log_view: u32, commit: u64, suffix: DvcSuffix| VsrAction::SendDoViewChange { + view: 1, + target: 1, + log_view, + op: 0, + commit, + group: GROUP, + suffix, + untrusted_log: true, + }; + let one_header = DvcSuffix::new(vec![PrepareHeader::default()], 0, 0); + for (vote, vouches_for) in [ + (do_view_change(5, true), "a head"), + (flagged(1, 0, DvcSuffix::empty()), "a log_view"), + (flagged(0, 1, DvcSuffix::empty()), "a commit point"), + (flagged(0, 0, one_header), "suffix headers"), + ] { + assert!( + !sendable_while_awaiting_transfer(&vote), + "a vote carrying {vouches_for} vouches for history" + ); + } assert!(!sendable_while_awaiting_transfer( &VsrAction::SendStartViewChange { view: 1, diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 93ea36b48..83cfc2382 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -629,7 +629,8 @@ impl ShardMetrics { /// 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. + /// waiting for a trusted quorum or 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)); @@ -929,7 +930,7 @@ impl ShardMetrics { 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", + once a trusted quorum or every replica reports", self.partitions_untrusted_log.clone(), ); registry.register( diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 38d3a1cff..472551404 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -3866,7 +3866,8 @@ mod tests { } } - /// Each replica's count of partitions waiting on every replica to report. + /// Each replica's count of partitions waiting on a trusted quorum or every + /// replica to report. fn untrusted_log_gauges(sim: &Simulator, namespace: IggyNamespace) -> Vec<i64> { sim.replicas .iter() @@ -3883,15 +3884,15 @@ mod tests { /// 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_losing(durability, crashed, &[]); } - /// [`restart_keeps_acknowledged_send`], with replica `lost` restarting - /// without the partition's log, so it may not elect on it. + /// [`restart_keeps_acknowledged_send`], with the replicas in `lost` + /// restarting without the partition's log, so they may not elect on it. fn restart_keeps_acknowledged_send_losing( durability: iggy_common::Durability, crashed: &[u8], - lost: Option<u8>, + lost: &[u8], ) { const PROGRESS_STEPS: usize = 4_000; const PAYLOAD: &[u8] = b"acknowledged-before-the-crash"; @@ -3925,7 +3926,7 @@ mod tests { sim.replica_crash(*replica); } for replica in crashed { - if lost == Some(*replica) { + if lost.contains(replica) { sim.replica_restart_losing_partitions(*replica, &[namespace]); } else { sim.replica_restart(*replica); @@ -3978,7 +3979,8 @@ mod tests { assert_eq!( gauge_saw_halt, durability == iggy_common::Durability::Replicated, - "{durability:?}: only a restarted replicated log waits on every replica" + "{durability:?}: only a restarted replicated log waits on a trusted quorum or every \ + replica" ); for _ in 0..PROGRESS_STEPS / 10 { sim.step(); @@ -4033,10 +4035,147 @@ mod tests { iggy_common::Durability::Replicated, iggy_common::Durability::Persisted, ] { - restart_keeps_acknowledged_send_losing(durability, &[0, 1, 2], Some(2)); + restart_keeps_acknowledged_send_losing(durability, &[0, 1, 2], &[2]); } } + /// Every replica crashes and two come back without the partition's log. + /// Both cast empty votes, broadcast to every peer as production sends a + /// `DoViewChange`, and must not answer each other's: the group settles on + /// the one log left. `replicated` only, since `persisted` cannot vouch for a + /// lone disk and rightly halts. + #[test] + fn given_every_replica_restarts_with_two_partition_logs_lost_should_settle_and_accept_sends() { + restart_keeps_acknowledged_send_losing( + iggy_common::Durability::Replicated, + &[0, 1, 2], + &[1, 2], + ); + } + + /// Every replica crashes and the replicas in `awaiting` come back with the + /// partition awaiting a state transfer. The shard lets only their empty votes + /// out, broadcast like any `DoViewChange`, and the others must still start a + /// view on the acknowledged send. Catching the awaiting replicas up takes a + /// partition state transfer, which in-memory storage does not model, so this + /// stops at the view. + fn full_restart_awaiting_transfer_starts_a_view(awaiting: &[u8]) { + const PROGRESS_STEPS: usize = 4_000; + 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, iggy_common::Durability::Replicated); + 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(b"acknowledged")]); + 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 old_view = sim + .partition_consensus_state(usize::from(primary), namespace) + .expect("primary hosts the partition") + .view; + + for replica in 0..3 { + sim.replica_crash(replica); + } + for replica in 0..3 { + if awaiting.contains(&replica) { + sim.replica_restart_losing_partitions(replica, &[namespace]); + } else { + sim.replica_restart(replica); + } + } + for replica in awaiting { + let partitions = sim.replicas[usize::from(*replica)] + .partition_shard(namespace) + .plane + .partitions(); + let partition = partitions + .get_mut_by_ns(&namespace) + .expect("a restarted replica rebuilds the partition"); + partition.force_requires_state_transfer_for_test(); + assert!(partition.requires_state_transfer()); + } + + let electing: Vec<u8> = (0..3) + .filter(|replica| !awaiting.contains(replica)) + .collect(); + let started = |sim: &Simulator| { + electing.iter().all(|replica| { + sim.replicas[usize::from(*replica)] + .partition_shard(namespace) + .plane + .partitions() + .get_by_ns(&namespace) + .is_some_and(|partition| { + let consensus = partition.consensus(); + consensus.status() == Status::Normal + && consensus.view() > old_view + && consensus.commit_max() >= committed_op + && !consensus.has_untrusted_log() + }) + }) + }; + assert!( + (0..PROGRESS_STEPS).any(|_| { + sim.step(); + started(&sim) + }), + "replicas {awaiting:?} await a state transfer and replicas {electing:?} never \ + started a view holding op {committed_op}: states={:?}", + (0..3) + .map(|replica| sim.partition_consensus_state(replica, namespace)) + .collect::<Vec<_>>() + ); + // The awaiting replicas stay `Recovering`, hearing each other's + // broadcast votes, so the vote traffic must die down with the view. + let do_view_changes = + |sim: &Simulator| sim.network.command_counts()[Command::DoViewChange as usize]; + let at_start = do_view_changes(&sim); + for _ in 0..PROGRESS_STEPS / 4 { + sim.step(); + } + let after_start = do_view_changes(&sim) - at_start; + assert!( + after_start <= 16, + "{after_start} DoViewChanges delivered after the view started" + ); + for replica in awaiting { + let state = sim + .partition_consensus_state(usize::from(*replica), namespace) + .expect("the awaiting replica hosts the partition"); + assert!( + !state.is_primary || state.status != Status::Normal, + "replica {replica} led a view while awaiting a state transfer" + ); + } + } + + #[test] + fn given_full_restart_when_one_replica_awaits_a_state_transfer_should_start_a_view_on_its_empty_vote() + { + full_restart_awaiting_transfer_starts_a_view(&[2]); + } + + #[test] + fn given_full_restart_when_two_replicas_await_a_state_transfer_should_start_a_view_on_their_empty_votes() + { + full_restart_awaiting_transfer_starts_a_view(&[1, 2]); + } + fn successful_send_reply_count(replies: &[Message<ReplyHeader>]) -> usize { replies .iter() diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs index d3474052b..bc4a2b4e0 100644 --- a/core/simulator/src/replica.rs +++ b/core/simulator/src/replica.rs @@ -20,7 +20,9 @@ use crate::deps::SimSuperblock; use crate::deps::{MemStorage, SimJournal, SimMuxStateMachine, SimSnapshot}; use configs::server::PersonalAccessTokenConfig; use configs::server::ServerConfig; -use consensus::{ClientTable, ConsensusClock, LocalPipeline, Sequencer, VsrConsensus, VsrState}; +use consensus::{ + ClientTable, ConsensusClock, JoinMode, LocalPipeline, Sequencer, VsrConsensus, VsrState, +}; use iggy_common::IggyByteSize; use iggy_common::variadic; use journal::Journal; @@ -275,10 +277,11 @@ pub fn new_shard( // backup. let restored_head = restored_op.unwrap_or(0); if !solo && (restored_head > 0 || recovered_state.is_some() || has_checkpoint) { - consensus.init_as_backup(); - consensus.begin_view_probe(); + consensus.join(JoinMode::ProbeAsBackup { + await_state_transfer: true, + }); } else { - consensus.init(); + consensus.join(JoinMode::Init); } if let (Some(journal), Some(head)) = (metadata_journal.as_ref(), restored_op) { let commit_watermark = journal.recovery_commit_watermark(solo).max(checkpoint_seq);
