This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch consensus-dvc-derived-nack in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e6c6c87c9af8fb5642ad764d10e2dea78353d41c Author: Krishna Vishal <[email protected]> AuthorDate: Wed Sep 9 23:31:59 2026 +0530 fix(consensus): count an unservable header as a nack --- core/consensus/src/dvc_merge.rs | 86 ++++++++++++++++++++++++++----------- core/consensus/src/plane_helpers.rs | 72 +++++++++++++++++++++++++++++-- core/simulator/src/lib.rs | 78 +++++++++++++++++++++++++++++++++ 3 files changed, 207 insertions(+), 29 deletions(-) diff --git a/core/consensus/src/dvc_merge.rs b/core/consensus/src/dvc_merge.rs index 516ef8ff8..0f8dfca4d 100644 --- a/core/consensus/src/dvc_merge.rs +++ b/core/consensus/src/dvc_merge.rs @@ -210,8 +210,9 @@ fn tally_op<'a>( }; let held = dvc.suffix.valid_header_at(index); + let offers_body = dvc.suffix.offers_body(index); if let (Some(held), Some(canonical)) = (held, canonical) - && dvc.suffix.offers_body(index) + && offers_body && held.checksum == canonical.checksum { copies += 1; @@ -221,21 +222,31 @@ fn tally_op<'a>( // Explicit: the sender proves it never prepared this op. nacks += 1; } else if let Some(held) = held { - // Only a sender BEHIND the canonical log_view can implicitly nack. - // Without this, corrupting one canonical header in transit turns every - // honest sender's correct header into an implicit nack against the - // garbage: a nack quorum on three replicas. A same-log_view - // disagreement is evidence, not a vote, and goes through `conflict`. - let may_nack_implicitly = dvc.log_view < canonical_log_view; - match canonical { - // Implicit: the sender holds a DIFFERENT prepare, so not this one. - Some(canonical) if may_nack_implicitly && held.checksum != canonical.checksum => { - nacks += 1; + if !offers_body && op > dvc.commit { + // A prepare is journaled before it is acked, so a header this + // sender cannot serve, above its own commit, proves it is outside + // the ack set. Below that commit a missing body is compaction. + nacks += 1; + } else { + // Only a sender BEHIND the canonical log_view can implicitly + // nack. Without this, corrupting one canonical header in transit + // turns every honest sender's correct header into an implicit + // nack against the garbage: a nack quorum on three replicas. A + // same-log_view disagreement is evidence, not a vote, and goes + // through `conflict`. + let may_nack_implicitly = dvc.log_view < canonical_log_view; + match canonical { + // Implicit: the sender holds a DIFFERENT prepare, so not this one. + Some(canonical) + if may_nack_implicitly && held.checksum != canonical.checksum => + { + nacks += 1; + } + // Implicit: no canonical sender holds anything here, so a newer + // view already truncated this op and the sender holds a corpse. + None if may_nack_implicitly => nacks += 1, + _ => {} } - // Implicit: no canonical sender holds anything here, so a newer - // view already truncated this op and the sender holds a corpse. - None if may_nack_implicitly => nacks += 1, - _ => {} } } } @@ -707,23 +718,24 @@ mod tests { } #[test] - fn given_header_without_a_servable_body_when_replicas_outstanding_should_await_repair() { - // Both senders have op 4's header, neither can serve its body, and replica 2 - // has not reported. A head whose body nobody holds would wedge the view. + fn given_header_without_a_servable_body_when_replicas_outstanding_should_truncate() { + // Neither sender journaled op 4, so the ack set was at most replica 2: + // short of a replication quorum, so it never committed. let mut quorum = dvc_quorum_array_empty(); let headers = suffix_headers(2, 4, 1); let header_only = DvcSuffix::new(headers, 0, 0b110); dvc_record(&mut quorum, dvc(0, 1, 4, 2, header_only.clone())); dvc_record(&mut quorum, dvc(1, 1, 4, 2, header_only)); - assert_eq!( - merge_dvc_quorum(&quorum, quorums_r3()), - MergeOutcome::AwaitingRepair { undecided_op: 4 } - ); + let MergeOutcome::Ready(log) = merge_dvc_quorum(&quorum, quorums_r3()) else { + panic!("two senders provably outside the ack set decide op 4"); + }; + assert_eq!(log.op_head, 3, "op 4 is truncated, not awaited"); + assert_eq!(log.commit_max, 2); } #[test] - fn given_all_replicas_reported_and_op_undecidable_should_deadlock() { + fn given_all_replicas_reported_and_no_body_should_truncate_rather_than_deadlock() { let mut quorum = dvc_quorum_array_empty(); let headers = suffix_headers(2, 4, 1); let header_only = DvcSuffix::new(headers, 0, 0b110); @@ -731,10 +743,32 @@ mod tests { dvc_record(&mut quorum, dvc(1, 1, 4, 2, header_only.clone())); dvc_record(&mut quorum, dvc(2, 1, 4, 2, header_only)); + let MergeOutcome::Ready(log) = merge_dvc_quorum(&quorum, quorums_r3()) else { + panic!("no replica can serve op 4 and all three say so"); + }; + assert_eq!(log.op_head, 3); + } + + #[test] + fn given_one_servable_copy_against_a_header_only_sender_should_keep_the_op() { + // Boundary: one servable copy outranks one derived nack. + let mut quorum = dvc_quorum_array_empty(); + let headers = suffix_headers(2, 4, 1); + dvc_record( + &mut quorum, + dvc(0, 1, 4, 2, DvcSuffix::new(headers.clone(), 0, 0b111)), + ); + dvc_record( + &mut quorum, + dvc(1, 1, 4, 2, DvcSuffix::new(headers, 0, 0b110)), + ); + + let MergeOutcome::Ready(log) = merge_dvc_quorum(&quorum, quorums_r3()) else { + panic!("op 4 is recoverable from replica 0, so the view must start"); + }; assert_eq!( - merge_dvc_quorum(&quorum, quorums_r3()), - MergeOutcome::Deadlocked { undecided_op: 4 }, - "with every replica in, an unrecoverable op stalls the view forever" + log.op_head, 4, + "a servable copy keeps the op against a single derived nack" ); } diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index a72995cf0..8074f1778 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -1437,6 +1437,41 @@ mod tests { (header, body) } + /// A DVC carrying `op` and `commit` only. Abstention: counts toward the + /// view-change quorum, says nothing about any op. + fn dvc_numbers_only( + replica: u8, + view: u32, + log_view: u32, + op: u64, + commit: u64, + ) -> (iggy_binary_protocol::DoViewChangeHeader, Body) { + use iggy_binary_protocol::DoViewChangeHeader; + + let headers: Vec<PrepareHeader> = Vec::new(); + let body = encode_body(&headers); + let header = DoViewChangeHeader { + checksum: 0, + checksum_body: 0, + cluster: 0, + size: u32::try_from(std::mem::size_of::<DoViewChangeHeader>() + body.len()) + .expect("synthetic DVC frame fits u32"), + view, + release: 0, + command: Command::DoViewChange, + replica, + reserved_frame: [0; 66], + op, + commit, + group: 0, + log_view, + reserved: [0; 68], + nack_bitset: 0, + present_bitset: 0, + }; + (header, body) + } + /// Headers for `low..=high`, high-to-low as a suffix requires, sealed and /// chained the way a real producer writes them. /// @@ -1500,6 +1535,38 @@ mod tests { consensus.set_local_dvc_suffix(crate::dvc_merge::suffix_all_present(headers)); } + /// The replica that prepared the head is gone for good, one survivor holds its + /// header without the body and the other never had it. Both survivors are + /// outside the ack set, so the op is truncated and the view starts. Counting the + /// header-only sender as neither copy nor nack instead waits for the crashed + /// replica forever. + #[test] + fn given_a_crashed_body_holder_when_merging_should_truncate_and_start_the_view() { + // View 3 of 3 replicas elects this one. + let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus, LocalPipeline::new()); + consensus.init(); + consensus.restore_commit_state(2, 2); + consensus.sequencer().set_sequence(4); + // Bit 0 is op 4: its header, never journaled. + let local = suffix_headers(2, 4, 0); + consensus.set_local_dvc_suffix(crate::view_change_quorum::DvcSuffix::new(local, 0, 0b110)); + + let _ = consensus.handle_start_view_change(PlaneKind::Metadata, &svc_header(1, 3)); + + // Replica 1 stops at op 3. Replica 2 held the only body and never reports. + let (dvc, body) = dvc_with_suffix(1, 3, 0, 3, 2, None); + let _ = consensus.handle_do_view_change(PlaneKind::Metadata, &dvc, &body); + + let pending = consensus + .pending_view_log() + .expect("two senders outside op 4's ack set decide it without replica 2"); + assert_eq!( + pending.op_head, 3, + "op 4 never committed, so it is truncated" + ); + assert_eq!(pending.commit_max, 2); + } + #[test] fn given_an_undecidable_quorum_when_a_later_dvc_decides_it_should_start_the_view() { // Reaching a view-change quorum is not the same as deciding a log. Latching @@ -1519,10 +1586,9 @@ mod tests { let _ = consensus.handle_start_view_change(PlaneKind::Metadata, &svc_header(1, 5)); - // Two peers report, reaching the quorum of 3. All three hold op 4's header, - // none can serve its body, and two replicas have yet to report. + // Two peers abstain, reaching the quorum of 3 and leaving op 4 one nack short. for replica in [1u8, 2] { - let (dvc, body) = dvc_with_suffix(replica, 5, 0, 4, 2, Some(4)); + let (dvc, body) = dvc_numbers_only(replica, 5, 0, 4, 2); let actions = consensus.handle_do_view_change(PlaneKind::Metadata, &dvc, &body); assert!(actions.is_empty()); } diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 51a2d3559..de7509577 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -4535,6 +4535,84 @@ mod tests { /// finds two replicas at the same op passes in silence, so `ops_compared` counts /// only ops witnessed on more than one replica, the subset that exercised the /// property. + /// A replica prepares an op and stays down. One survivor holds its header + /// without the body, the other never had it, and the merge must read that as + /// proof both are outside the ack set. Otherwise it waits for the crashed + /// replica: on this seed the cluster reached view 103 against `log_view` 3 with + /// both survivors caught up and one request retried 266 times unanswered. + /// + /// Asserts the drain, so it fails as the fuzzer does with the outstanding-request + /// report attached. + #[test] + fn view_change_completes_without_the_replica_that_prepared_the_head() { + use crate::workload::{ + self, FaultInjector, Workload, + options::{ActionWeights, WorkloadOptions}, + oracle, + }; + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + + let replica_count: u8 = 3; + let client_id: u128 = 1; + // `workload-fuzz --seed 211 --replicas 3 --ticks 4000 --plane metadata + // --journal-slots 80 --crash-prob 0.02 --restart-prob 0.08 --crash-primary`. + let seed = 211u64; + let root = tempfile::tempdir().expect("temp dir for the simulator's snapshots"); + let network_opts = packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed, + ..packet::PacketSimulatorOptions::default() + }; + // A bounded journal is what drives the WAL drain behind the header-only sender. + let mut sim = Simulator::with_checkpoints( + usize::from(replica_count), + std::iter::once(client_id), + network_opts, + false, + root.path(), + ); + sim.set_metadata_journal_slots(80); + + let ns = IggyNamespace::new(1, 1, 0); + sim.init_partition(ns); + let client = SimClient::new(client_id); + sim.register_client_with_primary(&client); + + let mut options = WorkloadOptions::new(seed, replica_count, vec![ns]); + options.client_count = 1; + options.crash_per_tick_ratio = 0.02; + options.restart_per_tick_ratio = 0.08; + options.spare_primary = false; + options.weights = ActionWeights::metadata_only(); + let mut workload = Workload::new(options); + + let clients = [client]; + let mut injector = FaultInjector::new(seed, replica_count); + let _ = workload::run_with_faults( + &mut sim, + &mut workload, + &clients, + 4_000, + u64::MAX, + &mut injector, + ); + + assert!( + injector.crashes() > 0, + "no replica crashed, so no view change ran and this proves nothing" + ); + assert!( + oracle::drive_to_quiesce(&mut sim, &mut workload, 50_000), + "{}", + oracle::quiesce_failure_report(&sim, &workload), + ); + } + #[test] fn committed_metadata_agrees_across_replicas() { use crate::workload::{
