This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/partition-routing in repository https://gitbox.apache.org/repos/asf/iggy.git
commit a3d28b3472845282a87849666cee11b2fcf882c2 Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Oct 14 11:44:03 2026 +0200 fix(server): close partition routing review gaps Repair gaps could discard committed history. Retain unknown entries under proof fences and remove only proven mismatches. Reconcile before repair admission and refresh StartView suffixes so restarted replicas and same-view retries recover safely. Bound forwarded HTTP holds and pin consumer progress and owner fences. Correct lifecycle coverage, avoid duplicate Go context lookups and document coordinated SDK/server upgrades. Extract the unrelated CI timing fallback into the separately preserved patch. Refs #4506 --- .github/workflows/pre-merge.yml | 16 +- core/consensus/src/impls.rs | 61 +++- core/consensus/src/plane_helpers.rs | 169 +++++---- core/journal/src/lib.rs | 13 +- core/journal/src/prepare_journal.rs | 75 +++- core/partitions/src/iggy_partition.rs | 165 +++++++-- core/partitions/src/journal.rs | 21 +- core/partitions/src/log.rs | 7 +- core/server/README.md | 16 + core/server/src/dispatch/mod.rs | 110 +++++- core/server/src/http/forward.rs | 8 + core/shard/src/lib.rs | 616 ++++++++++++++++---------------- core/shard/src/partition_io.rs | 258 +++++++++++-- core/simulator/src/bin/workload-fuzz.rs | 7 + core/simulator/src/deps.rs | 8 +- core/simulator/src/lib.rs | 242 ++++++------- core/simulator/src/workload/mod.rs | 59 +++ core/simulator/src/workload/shadow.rs | 13 +- foreign/go/client/tcp/tcp_core.go | 3 +- 19 files changed, 1243 insertions(+), 624 deletions(-) diff --git a/.github/workflows/pre-merge.yml b/.github/workflows/pre-merge.yml index 602563c75..24c0017f1 100644 --- a/.github/workflows/pre-merge.yml +++ b/.github/workflows/pre-merge.yml @@ -237,17 +237,11 @@ jobs: uses: actions/github-script@v9 with: script: | - let jobs; - try { - jobs = await github.rest.actions.listJobsForWorkflowRun({ - owner: context.repo.owner, - repo: context.repo.repo, - run_id: context.runId - }); - } catch (error) { - core.warning(`Job timings unavailable: ${error.message}`); - jobs = { data: { jobs: [] } }; - } + const jobs = await github.rest.actions.listJobsForWorkflowRun({ + owner: context.repo.owner, + repo: context.repo.repo, + run_id: context.runId + }); const jobTimes = {}; const jobStatuses = {}; diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 12dbb2a86..5a51303ae 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -1320,6 +1320,10 @@ where /// The installed view must prove the local suffix before it can be applied. /// Survives election resets until the plane successfully reconciles its journal. view_log_unproven: Cell<bool>, + /// The prefix below a repair gap is unknown; entries above it can be proven. + view_log_unproven_through: Cell<u64>, + /// Set only after adoption has removed the superseded tail. + view_log_missing_op: Cell<Option<u64>>, /// Tracks DVC messages received (only used by primary candidate) /// Stores metadata; actual log comes from message @@ -1681,6 +1685,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { journal_mutations: Cell::new(0), pending_view_log: RefCell::new(None), view_log_unproven: Cell::new(false), + view_log_unproven_through: Cell::new(u64::MAX), + view_log_missing_op: Cell::new(None), do_view_change_from_all_replicas: RefCell::new(dvc_quorum_array_empty()), do_view_change_quorum: Cell::new(false), sent_own_start_view_change: Cell::new(false), @@ -3545,10 +3551,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { } }; let headers = suffix.headers(); - if headers.is_empty() { + if headers.is_empty() && !self.view_log_unproven() { return header.op; } - *self.pending_view_log.borrow_mut() = Some(MergedLog { op_head: header.op, commit_max: header.commit, @@ -3644,7 +3649,10 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // drop the answer to its own probe, and its election would push the // discarded suffix back. let head = self.sequencer.current_sequence(); - if self.status.get() == Status::Normal && msg_view == self.log_view.get() && msg_op <= head + if self.status.get() == Status::Normal + && !self.view_log_unproven() + && msg_view == self.log_view.get() + && msg_op <= head { if msg_commit > self.commit_max() { self.advance_commit_max(msg_commit); @@ -3708,6 +3716,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { // Stale pipeline entries from the old view must be discarded self.view_log_unproven.set(needs_reconciliation); + self.view_log_unproven_through.set(u64::MAX); + self.view_log_missing_op.set(None); self.pipeline.borrow_mut().clear(); // Cross-check the announced head against the headers published with it: a @@ -3929,6 +3939,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.advance_commit_max(max_commit); *self.pending_view_log.borrow_mut() = Some(merged); self.view_log_unproven.set(true); + self.view_log_unproven_through.set(u64::MAX); + self.view_log_missing_op.set(None); // Stale pipeline entries are invalid in new view; reconciliation // replays from journal. @@ -3997,6 +4009,49 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { pub fn settle_view_log_proof(&self) { self.view_log_unproven.set(false); + self.invalidate_local_dvc_suffix(); + } + + pub fn fence_view_log(&self, op: u64, unproven_through: u64) { + self.view_log_unproven.set(true); + self.view_log_unproven_through.set(unproven_through); + self.view_log_missing_op.set(Some(op)); + self.invalidate_local_dvc_suffix(); + } + + #[must_use] + pub const fn view_log_entry_proven(&self, op: u64) -> bool { + !self.view_log_unproven() + || op <= self.commit_min() + || op > self.view_log_unproven_through.get() + } + + #[must_use] + pub const fn view_log_unproven_range(&self) -> std::ops::RangeInclusive<u64> { + self.commit_min().saturating_add(1)..=if self.view_log_unproven() { + self.view_log_unproven_through.get() + } else { + 0 + } + } + + #[must_use] + pub const fn view_log_repair_op(&self) -> Option<u64> { + if self.view_log_unproven() { + self.view_log_missing_op.get() + } else { + None + } + } + + pub fn note_repaired_view_header(&self, header: PrepareHeader) { + if let Some(pending) = self.pending_view_log.borrow_mut().as_mut() + && pending.headers.is_empty() + && pending.op_head == header.op + { + pending.headers.push(header); + self.invalidate_local_dvc_suffix(); + } } /// Whether a merge is parked, without cloning it. diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index afa16ce52..d541aca5e 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -853,7 +853,7 @@ where return false; } - if consensus.is_transferring() { + if consensus.is_transferring() || consensus.view_log_unproven() { return false; } @@ -903,37 +903,59 @@ where .await } -/// Find a held entry below a view's suffix that the new history does not prove. -/// -/// `canonical` is ordered by descending op. Entries above the local commit floor -/// survive only if their parent chain reaches its lowest header, or a prepare's -/// commit stamp proves them through the uninterrupted local chain. This runs once -/// per installed log, before repair starts filling any resulting gap. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ViewLogProof { + Proven, + /// Fetch `op`; retained entries above `unproven_through` match the view. + Missing { + op: u64, + unproven_through: u64, + }, + Divergent(u64), +} + +/// Prove held entries against the view's hash chain. Canonical headers can +/// bridge missing bodies; a gap with no header leaves its lower prefix unknown. #[must_use] -pub fn first_unproven_op( +pub fn view_log_proof( commit_floor: u64, canonical: &[PrepareHeader], op_head: u64, - last_op: Option<u64>, header_at: impl Fn(u64) -> Option<PrepareHeader>, -) -> Option<u64> { - let last_op = last_op?; +) -> ViewLogProof { let proven = locally_proven_commit(commit_floor, &header_at); - let (mut op, mut parent) = canonical.last().map_or((op_head, None), |lowest| { - (lowest.op.saturating_sub(1), Some(lowest.parent)) - }); - while op > proven { - match header_at(op) { - Some(local) - if local.checksum != CHECKSUM_UNSEALED && Some(local.checksum) == parent => - { - parent = Some(local.parent); - op -= 1; - } - _ => break, + if op_head <= proven { + return ViewLogProof::Proven; + } + let Some(head) = canonical.first().filter(|head| head.op == op_head) else { + return ViewLogProof::Missing { + op: op_head, + unproven_through: op_head, + }; + }; + let mut expected = head.checksum; + let mut missing = None; + for op in (proven + 1..=op_head).rev() { + let header = if let Some(local) = header_at(op) { + local + } else if let Some(header) = canonical.iter().find(|header| header.op == op) { + missing.get_or_insert(op); + *header + } else { + return ViewLogProof::Missing { + op, + unproven_through: op, + }; + }; + if header.checksum == CHECKSUM_UNSEALED || header.checksum != expected { + return ViewLogProof::Divergent(op); } + expected = header.parent; } - (proven.checked_add(1)?..=op.min(last_op)).find(|&held| header_at(held).is_some()) + missing.map_or(ViewLogProof::Proven, |op| ViewLogProof::Missing { + op, + unproven_through: 0, + }) } fn locally_proven_commit( @@ -2195,25 +2217,19 @@ mod tests { } #[test] - fn given_start_view_without_suffix_when_adopted_should_trust_the_announced_op() { - // Probe answers and stale-view corrections carry numbers only: a backup must - // still adopt, and record nothing it could mistake for the view's decision. + fn given_start_view_without_suffix_when_adopted_should_keep_proof_pending() { let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus, LocalPipeline::new()); consensus.init(); - let (header, body) = start_view_with_suffix(1, 1, 5, 3, false); assert!(body.is_empty()); let actions = consensus.handle_start_view(PlaneKind::Metadata, &header, &body); - assert!( - !actions.is_empty(), - "a numbers-only StartView must still adopt" - ); + assert!(!actions.is_empty()); assert_eq!(consensus.sequencer().current_sequence(), 5); - assert!( - consensus.pending_view_log().is_none(), - "no suffix means no canonical headers to verify against" - ); + let pending = consensus.pending_view_log().unwrap(); + assert_eq!(pending.op_head, 5); + assert_eq!(pending.headers, [] as [PrepareHeader; 0]); + assert!(consensus.view_log_unproven()); } #[test] @@ -2584,52 +2600,59 @@ mod tests { } #[test] - fn given_view_suffix_when_proving_local_history_should_keep_only_supported_entries() { - let chain = |ops: std::ops::RangeInclusive<u64>, mut parent, committed: u64| { - ops.map(|op| { - let header = PrepareHeader { - op, - parent, - commit: committed.min(op - 1), - checksum: u128::from(op), - ..Default::default() - }; - parent = header.checksum; - header + fn given_view_suffix_when_proving_local_history_should_distinguish_gaps_from_divergence() { + let local: Vec<_> = (3..=6) + .map(|op| PrepareHeader { + op, + parent: u128::from(op - 1), + commit: 2, + checksum: u128::from(op), + ..Default::default() }) - .collect::<Vec<_>>() - }; - let local = chain(3..=5, 2, 2); - let canonical = chain(6..=6, 5, 5); + .collect(); + let canonical = [*local.last().unwrap()]; let proof = |floor, entries: &[PrepareHeader], canonical: &[PrepareHeader]| { let held: BTreeMap<_, _> = entries.iter().map(|header| (header.op, *header)).collect(); - first_unproven_op(floor, canonical, 6, held.keys().last().copied(), |op| { - held.get(&op).copied() - }) + view_log_proof(floor, canonical, 6, |op| held.get(&op).copied()) }; - assert_eq!(proof(2, &local, &canonical), None, "matching parent chain"); + assert_eq!(proof(2, &local, &canonical), ViewLogProof::Proven); + let mut gap = local.clone(); + gap.remove(2); assert_eq!( - proof(2, &local[..2], &canonical), - Some(3), - "gap below suffix" + proof(2, &gap, &canonical), + ViewLogProof::Missing { + op: 5, + unproven_through: 5 + } ); - let mut stamped = local.clone(); - stamped[2].commit = 4; - assert_eq!(proof(2, &stamped, &[]), Some(5), "durable commit stamp"); - let mut replaced = local.clone(); - replaced[1].checksum += 1; + let bridged = [local[3], local[2]]; assert_eq!( - proof(2, &replaced, &canonical), - Some(3), - "broken parent link" + proof(2, &gap, &bridged), + ViewLogProof::Missing { + op: 5, + unproven_through: 0, + }, + "a canonical header proves the prefix even while its body is missing" ); - assert_eq!(proof(2, &local, &[]), Some(3), "empty canonical suffix"); - assert_eq!(proof(5, &local, &[]), None, "no uncommitted entries"); - stamped[1].checksum = CHECKSUM_UNSEALED; + gap[1].checksum += 10; + assert_eq!(proof(2, &gap, &bridged), ViewLogProof::Divergent(4)); assert_eq!( - proof(2, &stamped, &[]), - Some(3), - "unsealed chain proves no stamp" + proof(2, &gap, &canonical), + ViewLogProof::Missing { + op: 5, + unproven_through: 5 + }, + "a gap cannot prove the held prefix divergent" + ); + gap.insert(2, local[2]); + assert_eq!(proof(2, &gap, &canonical), ViewLogProof::Divergent(4)); + assert_eq!( + proof(2, &local, &[]), + ViewLogProof::Missing { + op: 6, + unproven_through: 6 + } ); + assert_eq!(proof(6, &local, &[]), ViewLogProof::Proven); } } diff --git a/core/journal/src/lib.rs b/core/journal/src/lib.rs index 823fce418..28342442b 100644 --- a/core/journal/src/lib.rs +++ b/core/journal/src/lib.rs @@ -56,13 +56,18 @@ pub trait Journal { /// it removed, which would mark the removed ops evictable when a suffix /// truncation needs them refillable. /// - /// Required, not defaulted: an `Unsupported` default hides a missing impl until - /// mid-view-change, where the caller can only wedge or start a view over a log it - /// cannot serve. + /// # Errors + /// I/O error if the rewrite fails. + fn truncate_from(&self, from_op: u64) -> impl Future<Output = io::Result<usize>> { + self.remove_range(from_op..=u64::MAX) + } + + /// Remove only the specified operations without advancing the snapshot floor. + /// View repair must preserve held entries on the other side of a gap. /// /// # Errors /// I/O error if the rewrite fails. - fn truncate_from(&self, from_op: u64) -> impl Future<Output = io::Result<usize>>; + fn remove_range(&self, ops: RangeInclusive<u64>) -> impl Future<Output = io::Result<usize>>; /// Highest op the index holds. Not derivable from [`Self::header`]: a caller /// looking for a suffix ABOVE some op has no bound to probe up to. diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs index 91a9d3f2c..d344d9580 100644 --- a/core/journal/src/prepare_journal.rs +++ b/core/journal/src/prepare_journal.rs @@ -953,31 +953,20 @@ impl Journal for PrepareJournal { self.last_op.get() } - /// Remove every entry at or above `from_op`, leaving the snapshot floor where - /// it is. Returns how many entries went. - /// - /// Deliberately not `drain`, which compacts a committed prefix and advances - /// `snapshot_op` past its range. Doing that to a suffix would declare everything - /// below the head snapshotted, letting `append` evict live entries repair cannot - /// put back, when those ops are exactly the ones that must stay refillable. - /// - /// For the one caller that needs it: a backup whose uncommitted entries disagree - /// with the log a view change settled on. They cannot be corrected in place, and - /// journal repair skips their ops as already-present, so dropping them is what - /// lets the primary's retransmission refill the range. + /// Preserve entries outside the repair range and leave removed ops refillable. /// /// # Errors /// I/O error if the rewrite fails. Past the rename the journal is poisoned on any /// failure, as in `drain`: serving a pre-truncation offset or appending at a stale - /// `write_offset` is worse than a hard stop. `from_op` must be at least 1. - async fn truncate_from(&self, from_op: u64) -> io::Result<usize> { + /// `write_offset` is worse than a hard stop. Ops are 1-based. + async fn remove_range(&self, ops: RangeInclusive<u64>) -> io::Result<usize> { if let Some(state) = self.poisoned.get() { return Err(Self::poisoned_io_error(state)); } - if from_op == 0 { + if *ops.start() == 0 { return Err(io::Error::new( io::ErrorKind::InvalidInput, - "truncate_from: ops are 1-based, so 0 would discard the whole journal", + "journal removal requires 1-based operations", )); } // Shares the drain guard: both rewrite the same WAL @@ -997,7 +986,7 @@ impl Journal for PrepareJournal { let offsets = self.offsets.borrow(); for slot in 0..self.slot_count { if let (Some(header), Some(offset)) = (&headers[slot], offsets[slot]) { - if header.op >= from_op { + if ops.contains(&header.op) { removed += 1; } else { live.push((*header, offset)); @@ -1024,7 +1013,7 @@ impl Journal for PrepareJournal { } for slot in 0..self.slot_count { if let Some(header) = &headers[slot] - && header.op >= from_op + && ops.contains(&header.op) { headers[slot] = None; offsets[slot] = None; @@ -1463,6 +1452,56 @@ mod tests { assert!(journal.header(4).is_some()); } + #[compio::test] + async fn given_a_repair_gap_when_rewriting_and_reopening_should_preserve_and_refill_both_sides() + { + const HEAD: u64 = 5; + const GAP: u64 = 3; + const BODY: usize = 64; + let dir = tempdir().unwrap(); + let path = dir.path().join("journal.wal"); + let mut parent = 0; + let entries: Vec<_> = (1..=HEAD) + .map(|op| { + let entry = make_identity_sealed_prepare(op, BODY, parent); + parent = entry.header().checksum; + entry + }) + .collect(); + { + let journal = PrepareJournal::open(&path, 0).await.unwrap(); + for entry in &entries { + journal.append(entry.deep_copy()).await.unwrap(); + } + assert_eq!(journal.remove_range(GAP..=GAP).await.unwrap(), 1); + } + { + let journal = PrepareJournal::open(&path, 0).await.unwrap(); + assert_eq!(journal.snapshot_op(), 0); + assert_eq!(journal.last_op(), Some(HEAD)); + assert!(journal.header(GAP as usize).is_none()); + for entry in entries.iter().filter(|entry| entry.header().op != GAP) { + let header = *journal.header(entry.header().op as usize).unwrap(); + assert_eq!( + journal.entry(&header).await.unwrap().as_slice(), + entry.as_slice() + ); + } + journal + .append(entries[GAP as usize - 1].deep_copy()) + .await + .unwrap(); + } + let journal = PrepareJournal::open(&path, 0).await.unwrap(); + for entry in &entries { + let header = *journal.header(entry.header().op as usize).unwrap(); + assert_eq!( + journal.entry(&header).await.unwrap().as_slice(), + entry.as_slice() + ); + } + } + #[compio::test] async fn truncate_from_survives_reopen() { // The rewrite has to be durable, not just reflected in the index. diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 601d58ac2..82f8d5313 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -528,7 +528,7 @@ pub enum PendingPartitionTransition { drain: Option<crate::PersistenceDrain>, }, Truncate { - from_op: u64, + ops: std::ops::RangeInclusive<u64>, drain: Option<crate::PersistenceDrain>, actions: Vec<consensus::VsrAction>, peer: Option<u8>, @@ -2344,7 +2344,10 @@ where return; }; while let Some(op) = persistence.head().checked_add(1) { - if op > through || op > self.consensus.sequencer().current_sequence() { + if op > through + || op > self.consensus.sequencer().current_sequence() + || !self.consensus.view_log_entry_proven(op) + { break; } let Some(prepare) = self.log.journal().inner.repair_entry(op) else { @@ -5607,12 +5610,12 @@ where pub fn defer_view_truncation( &mut self, - from_op: u64, + ops: std::ops::RangeInclusive<u64>, actions: Vec<consensus::VsrAction>, peer: Option<u8>, ) { self.transition = Some(PendingPartitionTransition::Truncate { - from_op, + ops, drain: None, actions, peer, @@ -5818,12 +5821,11 @@ where let message = self.deferred_views.swap_remove(next); return Ok(crate::PartitionIoStep::Transition(message)); } - if let Some(PendingPartitionTransition::Truncate { from_op, drain, .. }) = - &mut self.transition + if let Some(PendingPartitionTransition::Truncate { ops, drain, .. }) = &mut self.transition { if let Some(persistence) = &self.persistence { if drain.is_none() { - persistence.truncate_from(*from_op); + persistence.truncate_from(*ops.start()); if persistence.start() { self.consensus .message_bus() @@ -5841,15 +5843,12 @@ where persistence.finish_drain(drain); } let Some(PendingPartitionTransition::Truncate { - from_op, - actions, - peer, - .. + ops, actions, peer, .. }) = self.transition.take() else { unreachable!("truncate transition exists"); }; - self.truncate_resident_from(from_op).await?; + self.remove_resident_range(ops).await?; return Ok(crate::PartitionIoStep::ViewApplied { actions, peer }); } if matches!( @@ -8765,11 +8764,45 @@ where } async fn truncate_resident_from(&mut self, from_op: u64) -> Result<usize, IggyError> { + self.remove_resident_range(from_op..=u64::MAX).await + } + + /// Replace a proven mismatch while retaining the canonical tail across gaps. + /// + /// # Errors + /// Returns an error if the WAL or resident indexes cannot be rewritten. + pub async fn remove_divergent_prepare(&mut self, op: u64) -> Result<(), IggyError> { + if op + <= self + .consensus + .commit_min() + .max(self.log.journal().inner.evicted_through()) + { + return Err(IggyError::InvalidCommand); + } + if let Some(persistence) = &self.persistence { + // The durable WAL is contiguous. A canonical tail beyond a divergent + // entry is repair-held and must be persisted again after the gap fills. + persistence.truncate_from(op); + self.start_persistence(); + persistence + .drain_with_timeout() + .await + .map_err(|_| IggyError::CannotSyncFile)?; + } + self.remove_resident_range(op..=op).await?; + Ok(()) + } + + async fn remove_resident_range( + &mut self, + ops: std::ops::RangeInclusive<u64>, + ) -> Result<usize, IggyError> { let mut entries = self.log.journal().inner.resident_entries(); entries.sort_unstable_by_key(peek_op); let mut retained_info = JournalInfo::default(); let mut retained_next_offset = 0; - let mut rewind_next_offset = None; + let mut removed_messages = false; for entry in &entries { if peek_operation(entry) != Operation::SendMessages { continue; @@ -8779,12 +8812,8 @@ where if batch.message_count() == 0 { continue; } - if peek_op(entry) >= from_op { - rewind_next_offset = Some( - rewind_next_offset.map_or(batch.header.base_offset, |offset: u64| { - offset.min(batch.header.base_offset) - }), - ); + if ops.contains(&peek_op(entry)) { + removed_messages = true; continue; } accumulate_committed_info( @@ -8823,14 +8852,14 @@ where .log .journal() .inner - .truncate_from(from_op) + .remove_range(ops.clone()) .await .map_err(|_| IggyError::CannotAppendMessage)?; self.log.journal_mut().info = retained_info; self.log.active_segment_mut().current_position = restored_position; - if let Some(next_offset) = rewind_next_offset { - let next_offset = next_offset.max(minimum_next_offset); + if removed_messages { + let next_offset = minimum_next_offset; self.dirty_offset .store(next_offset.saturating_sub(1), Ordering::Relaxed); // The APPEND bit follows the rewound counter. The committed bit does @@ -8844,9 +8873,8 @@ where } } self.consensus.note_journal_mutation(); - let commit_max = self.consensus.commit_max(); self.pending_consumer_offset_commits - .retain(|op, _| *op < from_op || *op <= commit_max); + .retain(|op, _| !ops.contains(op)); self.offset_reservations_need_resync.set(true); self.truncated_since_scan.set(true); Ok(removed) @@ -11592,7 +11620,7 @@ where /// commit walk runs at `RepairDone`, after the floor is known. pub async fn apply_repaired_prepare(&mut self, message: Message<PrepareHeader>) { let header = *message.header(); - if !self.prepare_history_available(&header) { + if self.view_truncation_pending() || !self.prepare_history_available(&header) { return; } let Some(session) = self.repair else { @@ -11627,7 +11655,10 @@ where // A held frame is no progress: a peer that re-serves the same chunk // every round must still spend the budget, or it cannot be rotated // away from. - if self.log.journal().inner.holds_op(header.op) { + if let Some(held) = self.log.journal().inner.header_by_op(header.op) { + if held.checksum == header.checksum { + self.consensus.note_repaired_view_header(header); + } return; } // Any other in-window frame proves the stream is alive; only silence @@ -11667,6 +11698,7 @@ where // its peer serves it again every round, and a refund per serve would // pin the session to that peer. self.restart_repair_budget(); + self.consensus.note_repaired_view_header(header); self.persist_repaired_prefix(); // Advance the sequencer only along the CONTIGUOUS journaled // frontier. DVC advertises `op = sequencer.current_sequence()` and @@ -14317,6 +14349,37 @@ mod tests { assert_eq!(partition.log.journal().inner.resident_count(), 0); } + #[compio::test] + async fn given_a_discarded_suffix_with_an_offset_gap_when_truncated_should_reuse_the_retained_frontier() + { + let mut partition = partition_at_view(0, 0); + let retained = checksummed_segment_prepare(1, 0, 0, b"retained"); + let parent = retained.header().checksum; + partition + .append_repaired_send_messages(retained) + .await + .unwrap(); + let discarded = checksummed_segment_prepare(3, parent, 2, b"discarded"); + partition + .append_repaired_send_messages(discarded) + .await + .unwrap(); + partition.consensus.sequencer().set_sequence(3); + + partition.truncate_uncommitted_from(2).await.unwrap(); + + assert_eq!(partition.dirty_offset.load(Ordering::Relaxed), 0); + let replacement = checksummed_segment_prepare(2, parent, 1, b"replacement"); + partition + .append_received_send_messages_to_journal(replacement) + .await + .unwrap(); + assert_eq!(partition.dirty_offset.load(Ordering::Relaxed), 1); + assert!(partition.log.journal().inner.holds_op(1)); + assert!(partition.log.journal().inner.holds_op(2)); + assert!(!partition.log.journal().inner.holds_op(3)); + } + #[compio::test] async fn live_wal_rollback_preserves_replacement_indexes_and_pollable_bodies() { let root = tempfile::tempdir().unwrap(); @@ -21574,15 +21637,15 @@ mod tests { crate::PartitionIoStep::Transition(server_common::MessageBag::StartViewChange(message)) if message.header().view == 2)); partition.transition = Some(PendingPartitionTransition::Truncate { - from_op: 3, + ops: 3..=u64::MAX, actions: Vec::new(), peer: None, drain: None, }); partition.defer_view_transition(start_view(3)); assert!(matches!( - partition.transition, - Some(PendingPartitionTransition::Truncate { from_op: 3, .. }) + partition.transition.as_ref(), + Some(PendingPartitionTransition::Truncate { ops, .. }) if *ops == (3..=u64::MAX) )); partition.defer_view_transition(start_view(4)); partition.transition = None; @@ -24171,6 +24234,48 @@ mod tests { assert!(partition.fatal().is_none(), "{:?}", partition.fatal()); } + #[compio::test] + async fn given_queued_auto_commit_when_its_owner_is_replaced_should_refuse_it_at_promotion() { + const GROUP_ID: u32 = 7; + const CLIENT_ID: u128 = 42; + let mut partition = test_partition(); + partition.stats.increment_messages_count(1); + install_test_owner(&mut partition, u64::from(GROUP_ID), CLIENT_ID).await; + let store = partition + .build_poll_auto_commit_request(ConsumerKind::ConsumerGroup, GROUP_ID, 0) + .unwrap(); + let generation = store.header().owner_generation; + install_test_owner(&mut partition, u64::from(GROUP_ID), CLIENT_ID + 1).await; + assert!( + partition + .consumer_group_owner(u64::from(GROUP_ID)) + .unwrap() + .generation + > generation + ); + let head = partition.consensus.sequencer().current_sequence(); + assert!( + partition + .consensus + .push_queued_request(consensus::RequestEntry::with_sender(store, None),) + .is_ok() + ); + + partition.drain_request_queue_into_prepares(1).await; + + assert_eq!( + partition.consensus.sequencer().current_sequence(), + head, + "an internal auto-commit must retain the owner generation it captured" + ); + assert!(partition.consensus.pop_queued_request().is_none()); + assert_eq!( + partition.group_offset_state(u64::from(GROUP_ID)), + (None, None) + ); + assert!(partition.fatal().is_none(), "{:?}", partition.fatal()); + } + #[compio::test] async fn given_queued_capacity_denial_when_promoting_should_use_the_slot_for_next_existing_key() { @@ -24271,7 +24376,7 @@ mod tests { partition.on_ack(stale_ack, &config).await; assert_eq!(partition.pending_owner_install.get(), Some(discarded)); - partition.defer_view_truncation(discarded_op, Vec::new(), None); + partition.defer_view_truncation(discarded_op..=u64::MAX, Vec::new(), None); assert!(matches!( partition.resume_io(&config).await, crate::PartitionIoStep::ViewApplied { .. } diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index 98d63e63e..03bbd5060 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -1127,19 +1127,12 @@ impl Journal for PartitionJournal<PartitionJournalMemStorage> { op_to_storage_offset.last_key_value().map(|(op, _)| *op) } - /// Drop every entry at or above `from_op`, rebuilding the indexes. Same - /// drain-and-re-append shape as `evict_prefix`, from the other end and retaining - /// nothing: `append` has no slot-collision check here, so a superseded entry left - /// in place sits beside the new view's prepare at the same op and - /// positional materialization flushes the stale one. - /// - /// Dropped entries do NOT enter the evicted repair ring: it answers repair for - /// committed ops, and these are ones the view just decided against. - async fn truncate_from(&self, from_op: u64) -> io::Result<usize> { - if from_op == 0 { + /// Removed entries must stay refillable and cannot enter the committed repair ring. + async fn remove_range(&self, ops: std::ops::RangeInclusive<u64>) -> io::Result<usize> { + if *ops.start() == 0 { return Err(io::Error::new( io::ErrorKind::InvalidInput, - "truncate_from: ops are 1-based, so 0 would discard the whole journal", + "journal removal requires 1-based operations", )); } let all_entries = { @@ -1148,15 +1141,15 @@ impl Journal for PartitionJournal<PartitionJournalMemStorage> { }; // Positional against `headers` until the clear below (see the length-lock // invariant on `append_with_meta_sync`), so the ops are captured first. - let ops: Vec<u64> = { + let entry_ops: Vec<u64> = { let headers = unsafe { &*self.headers.get() }; headers.iter().map(|header| header.op).collect() }; self.clear_resident_indexes(); let mut removed = 0usize; - for (op, entry) in ops.into_iter().zip(all_entries) { - if op >= from_op { + for (op, entry) in entry_ops.into_iter().zip(all_entries) { + if ops.contains(&op) { removed += 1; continue; } diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs index 6cab05c49..da36addce 100644 --- a/core/partitions/src/log.rs +++ b/core/partitions/src/log.rs @@ -112,8 +112,11 @@ where self.inner.drain(ops) } - fn truncate_from(&self, from_op: u64) -> impl Future<Output = std::io::Result<usize>> { - self.inner.truncate_from(from_op) + fn remove_range( + &self, + ops: std::ops::RangeInclusive<u64>, + ) -> impl Future<Output = std::io::Result<usize>> { + self.inner.remove_range(ops) } fn last_op(&self) -> Option<u64> { diff --git a/core/server/README.md b/core/server/README.md index c2f859095..3631ae23e 100644 --- a/core/server/README.md +++ b/core/server/README.md @@ -39,6 +39,22 @@ cargo run --bin iggy-server --release -- --replica-id 0 Command-line arguments include `--replica-id`, `--fresh` (`-f`), `--with-default-root-credentials`, and `--list-config-env-vars`. Other settings come from configuration. +## Upgrading + +Deploy matching server and SDK releases together. Rolling upgrades and mixed +server versions are unsupported. Partition control frames from older servers +carry incarnation zero and are rejected, so a mixed-version cluster cannot +complete partition view changes. The `IGGY-OWNERSHIP-HISTORY-5` storage gate also +rejects older data directories without modifying them. Existing data requires a +separately validated offline migration or restore procedure before this release +can use it. + +Cluster metadata includes `replica_id`. Java's `ClusterNode` record adds a leading +`replicaId` component, changing constructor calls and record patterns. C# callers +must initialize the required `ReplicaId` property. The CLI's cluster metadata list +appends a replica ID column. Go, Java, Node.js and C# clients use command 105's new +partition routes and refresh on `NotPrimary` (89); use their matching releases. + ## Configuration Settings are read from [config.toml](config.toml), resolved relative to the working directory. Set `IGGY_CONFIG_PATH` to load a different file. diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs index b4776fefe..0646ffdca 100644 --- a/core/server/src/dispatch/mod.rs +++ b/core/server/src/dispatch/mod.rs @@ -1167,7 +1167,9 @@ mod tests { POLL_MESSAGES_ON_PRIMARY_CODE, }; use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy; - use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; + use iggy_binary_protocol::requests::consumer_offsets::{ + GetConsumerOffsetRequest, StoreConsumerOffsetRequest, + }; use iggy_binary_protocol::requests::messages::PollMessagesRequest; use iggy_binary_protocol::requests::streams::CreateStreamRequest; use iggy_binary_protocol::requests::system::BindSessionRequest; @@ -2486,6 +2488,112 @@ mod tests { } } + #[compio::test] + async fn given_a_held_store_when_next_and_offset_reads_queue_should_wait_for_its_guard() { + const CONSUMER_ID: u32 = 1; + const NEXT_REQUEST: u64 = 2; + const OFFSET_REQUEST: u64 = 3; + let (bus, shard, host) = root_host(ServerConfig::default()); + let namespace = create_one_partition(&shard); + bind_root_session(&shard, host.sessions()).await; + let config = Arc::new(ServerConfig::default()); + let drain = Rc::new(ClientDrain::new(&config)); + let store = StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(CONSUMER_ID)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + offset: 0, + ack: AckLevel::Quorum, + } + .to_bytes(); + let store = wire_request( + Operation::StoreConsumerOffset, + ROOT_CLIENT, + ROOT_SESSION, + 1, + &store, + ) + .into_generic(); + let key = progress_key(&shard, &store).expect("store must take a consumer progress lane"); + assert_eq!(key.0, namespace.inner()); + let held = drain.admit_progress(key); + let poll = PollMessagesRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(CONSUMER_ID)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + strategy: WirePollingStrategy::next(), + count: 1, + auto_commit: false, + } + .to_bytes(); + let offset = GetConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(CONSUMER_ID)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + } + .to_bytes(); + let mut reads = VecDeque::new(); + for (request, code, body) in [ + (NEXT_REQUEST, POLL_MESSAGES_CODE, poll), + (OFFSET_REQUEST, GET_CONSUMER_OFFSET_CODE, offset), + ] { + let mut message = wire_request( + Operation::NonReplicated, + ROOT_CLIENT, + ROOT_SESSION, + request, + &body, + ); + bytemuck::checked::from_bytes_mut::<RequestHeader>( + &mut message.as_mut_slice()[..size_of::<RequestHeader>()], + ) + .reserved[..4] + .copy_from_slice(&code.to_le_bytes()); + reads.push_back(QueuedClientRequest::from(message.into_generic())); + } + let queues = Rc::new(RefCell::new(AHashMap::from_iter([(ROOT_TRANSPORT, reads)]))); + let running = drain_client_requests( + Rc::clone(&shard), + Rc::clone(host.sessions()), + config, + 1, + Rc::clone(&queues), + &drain, + ROOT_TRANSPORT, + ); + futures::pin_mut!(running); + assert!(futures::poll!(running.as_mut()).is_pending()); + settle().await; + assert_eq!( + queues + .borrow() + .get(&ROOT_TRANSPORT) + .map_or(0, VecDeque::len), + 2, + "Next without auto-commit and GetConsumerOffset must both stay queued" + ); + assert!(bus.client_replies.borrow().is_empty()); + + drop(held); + assert!(futures::poll!(running.as_mut()).is_pending()); + await_replies(&bus, 1).await; + while futures::poll!(running.as_mut()).is_pending() { + settle().await; + } + await_replies(&bus, 2).await; + let replies = bus.client_replies.borrow(); + assert_eq!( + replies + .iter() + .map(|(_, reply)| reply_request(reply)) + .collect::<Vec<_>>(), + [NEXT_REQUEST, OFFSET_REQUEST] + ); + } + #[compio::test] async fn given_queued_progress_when_metadata_changes_should_resolve_once_per_frontier() { const CREATED_FRONTIER: u64 = ROOT_SESSION + 1; diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs index 6046bd511..52c438b4d 100644 --- a/core/server/src/http/forward.rs +++ b/core/server/src/http/forward.rs @@ -551,6 +551,14 @@ async fn attach_partition_request(state: HttpState, request: Request, next: Next if let Err(rejection) = resolve_credential(&state, bearer).await { return rejection.into_response(); } + let _guard = if matches!(hop, PartitionHop::Target(_)) { + let Some(guard) = ForwardGuard::admit(&state.forward.in_flight) else { + return forward_busy(); + }; + Some(guard) + } else { + None + }; let (mut parts, request_body) = request.into_parts(); let Ok(body) = to_bytes(request_body, state.forward.body_limit).await else { return error_response( diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 7215ba796..e7e76a925 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -4613,7 +4613,7 @@ where &self, partition: &IggyPartition<B, SB>, partition_incarnation: u64, - frame: &'static str, + frame: Command, ) -> bool { if partition_incarnation == partition.created_revision() { return true; @@ -4621,7 +4621,7 @@ where tracing::debug!( shard = self.id, namespace = partition.consensus().group(), - frame, + ?frame, partition_incarnation, hosted_incarnation = partition.created_revision(), "dropping partition control frame of another incarnation" @@ -4643,7 +4643,7 @@ where partition_incarnation: u64, view: u32, replica: u8, - frame: &'static str, + frame: Command, ) -> Option<&'a mut IggyPartition<B, SB>> where B: MessageBus, @@ -4654,7 +4654,7 @@ where namespace, view, replica, - frame, + ?frame, "dropping VSR control frame: namespace matches neither metadata nor partition consensus" ); return None; @@ -4667,7 +4667,8 @@ where if !self.partition_incarnation_matches(partition, partition_incarnation, frame) { return None; } - if view < partition.consensus().view() { + // A stale primary's heartbeat asks the current primary to announce its view. + if view < partition.consensus().view() && frame != Command::Commit { return None; } Some(partition) @@ -4706,7 +4707,7 @@ where header.partition_incarnation, header.view, header.replica, - "StartViewChange", + Command::StartViewChange, ) else { return; }; @@ -4779,7 +4780,7 @@ where header.partition_incarnation, header.view, header.replica, - "DoViewChange", + Command::DoViewChange, ) else { return; }; @@ -4843,13 +4844,15 @@ where }; let actions = consensus.handle_start_view(PlaneKind::Metadata, &header, suffix_body); let adopted = !actions.is_empty(); - if consensus.view_log_unproven() { + if !consensus.is_transferring() && consensus.view_log_unproven() { // First chance to spot a local entry disagreeing with the view's log. // Ahead of the local dispatch below: it truncates the journal that // `RebuildPipeline` reads back, so a rebuild before it would seed the // pipeline from the entries this is about to drop. self.reconcile_metadata_view_divergence().await; if consensus.view_log_unproven() { + self.maybe_request_metadata_repair(consensus, header.replica) + .await; return; } } @@ -4926,7 +4929,7 @@ where header.partition_incarnation, header.view, header.replica, - "StartView", + Command::StartView, ) else { return; }; @@ -4948,7 +4951,7 @@ where partition .consensus() .handle_start_view(PlaneKind::Partitions, &header, suffix_body); - if partition.consensus().view_log_unproven() { + if !partition.consensus().is_transferring() && partition.consensus().view_log_unproven() { // Any stream armed before this adoption belongs to the superseded // view. Repair bodies carry no nonce, so drop the receiving session // before reconciling or arming the new view's canonical range. @@ -4959,14 +4962,15 @@ where // the relics above the adopted head. let pending = partition.consensus().pending_view_log(); if partition.has_io_dispatcher() - && let Some(from_op) = - partition_view_divergence_from(self.id, partition, pending.as_ref()) + && let Some(from_op) = partition_view_divergence_from(partition, pending.as_ref()) { - partition.defer_view_truncation(from_op, actions, Some(header.replica)); + partition.defer_view_truncation(from_op..=u64::MAX, actions, Some(header.replica)); return; } reconcile_partition_view_divergence(self.id, partition, pending.as_ref()).await; if partition.consensus().view_log_unproven() { + self.maybe_request_partition_repair(partition, header.replica) + .await; return; } } @@ -5085,6 +5089,7 @@ where // dispatch gate does. Withhold on failure; the stale peer keeps // heartbeating, so it re-triggers once the tick persists. if planes.0.persist_superblock_if_needed(consensus).await { + refresh_metadata_dvc_suffix(consensus, planes.0.journal.as_ref()); respond_start_view::<B, _, MJ>(consensus).await; } } @@ -5100,7 +5105,7 @@ where header.partition_incarnation, header.view, header.replica, - "Commit", + Command::Commit, ) else { return; }; @@ -5128,6 +5133,7 @@ where if !partition.requires_state_transfer() && partition.persist_superblock_if_needed().await { + refresh_partition_dvc_suffix(partition); respond_start_view::<B, _, MJ>(consensus).await; } } @@ -5171,7 +5177,7 @@ where if !self.partition_incarnation_matches( partition, header.partition_incarnation, - "RequestStartView", + Command::RequestStartView, ) { return; } @@ -5224,8 +5230,8 @@ where // Served in `ViewChange` too: the replicas holding a missing body are // exactly those in `ViewChange`, so refusing would deadlock the repair // the new primary waits on. Read-only; the requester decides. - if consensus.is_transferring() || consensus.view_log_unproven() { - // Transfer and unresolved view reconciliation can replace these entries. + if consensus.is_transferring() { + // Transfer can replace these entries. return; } if !matches!(consensus.status(), Status::Normal | Status::ViewChange) { @@ -5252,6 +5258,9 @@ where // the pump against a 10 ms tick. Nothing at or below the watermark is // servable anyway. The partition arm jumps to `retained_from` likewise. let mut from_op = header.from_op.max(journal.snapshot_op() + 1); + if !consensus.view_log_entry_proven(from_op) { + return; + } #[allow(clippy::cast_possible_truncation)] while from_op <= to_op && journal.header(from_op as usize).is_none() { from_op += 1; @@ -5301,6 +5310,9 @@ where let chunk_end = to_op.min(from_op.saturating_add(repair_chunk_max - 1)); let mut served_through = from_op.saturating_sub(1); for op in from_op..=chunk_end { + if !consensus.view_log_entry_proven(op) { + break; + } #[allow(clippy::cast_possible_truncation)] let Some(entry_header) = journal.header(op as usize).map(|h| *h) else { break; @@ -5336,13 +5348,14 @@ where if !self.partition_incarnation_matches( partition, header.partition_incarnation, - "RequestPrepares", + Command::RequestPrepares, ) { return; } - if !partition.consensus().is_normal() - || partition.consensus().view_log_unproven() - || partition.view_truncation_pending() + if !matches!( + partition.consensus().status(), + Status::Normal | Status::ViewChange + ) || partition.view_truncation_pending() { return; } @@ -5377,6 +5390,9 @@ where .repair_retained_from() .unwrap_or_else(|| partition.consensus().commit_min().saturating_add(1)); let mut from_op = header.from_op; + if !partition.consensus().view_log_entry_proven(from_op) { + return; + } if retained_from > from_op { self.send_repair_range_reply( cluster, @@ -5394,6 +5410,9 @@ where let chunk_end = to_op.min(from_op.saturating_add(repair_chunk_max - 1)); let mut served_through = from_op.saturating_sub(1); for op in from_op..=chunk_end { + if !partition.consensus().view_log_entry_proven(op) { + break; + } let Some(entry) = partition.log.journal().inner.repair_entry(op) else { break; }; @@ -5547,6 +5566,13 @@ where let journal = journal.handle(); #[allow(clippy::cast_possible_truncation)] if journal.header(header.op as usize).is_some() { + if journal + .header(header.op as usize) + .is_some_and(|held| held.checksum == header.checksum) + { + consensus.note_repaired_view_header(header); + self.reconcile_metadata_view_divergence().await; + } return; } if let Err(error) = journal.append(msg).await { @@ -5566,6 +5592,7 @@ where // journaled it, one nack away from truncating an op it is about to // acknowledge. consensus.note_journal_mutation(); + consensus.note_repaired_view_header(header); // Contiguous-frontier advance, mirroring // `apply_repaired_prepare`: DVC advertises the sequencer, so a // hole below a repaired op must stall the advance rather than @@ -5579,6 +5606,7 @@ where consensus.sequencer().set_sequence(frontier); consensus.set_last_prepare_checksum(frontier_checksum); } + self.reconcile_metadata_view_divergence().await; return; } // A metadata-plane op that did not match above (no metadata consensus on @@ -5617,6 +5645,8 @@ where return; } partition.apply_repaired_prepare(msg).await; + let pending = partition.consensus().pending_view_log(); + reconcile_partition_view_divergence(self.id, partition, pending.as_ref()).await; } /// Repair stream terminator: `RangeEvicted` settles the partition commit @@ -5657,6 +5687,9 @@ where // below `served_through` is left to the retry timer. let commit_min = consensus.commit_min(); let done = commit_min >= session.to_op; + let proof_advanced = consensus + .view_log_repair_op() + .is_some_and(|op| op < session.from_op); tracing::info!( shard = self.id, through_op = header.op, @@ -5669,18 +5702,22 @@ where // Gating on the walk and not the frame keeps a peer that // terminates every round while serving nothing useful from // clearing its own budget and becoming un-rotatable. - if done || commit_min > before { + if done || commit_min > before || proof_advanced { self.note_metadata_repair_walked(); } if done { *self.metadata_repair.borrow_mut() = None; - } else if repair_chunk_walked(before, commit_min, header.op) { + } else if proof_advanced || repair_chunk_walked(before, commit_min, header.op) { + let from_op = consensus.view_log_repair_op().unwrap_or(commit_min + 1); + if let Some(live) = self.metadata_repair.borrow_mut().as_mut() { + live.from_op = from_op; + } self.send_request_prepares( consensus.cluster(), consensus.replica(), session.peer, session.nonce, - commit_min + 1, + from_op, session.to_op, header.group, consensus.partition_incarnation(), @@ -5838,7 +5875,7 @@ where if !self.partition_incarnation_matches( partition, header.partition_incarnation, - "RepairRangeReply", + header.command, ) { return; } @@ -6103,19 +6140,23 @@ where // 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 // adoption to hang it off. - if let Some(from_op) = - partition_view_divergence_from(self.id, partition, Some(&pending)) + if partition.has_io_dispatcher() + && let Some(from_op) = partition_view_divergence_from(partition, Some(&pending)) { - if partition.has_io_dispatcher() { - partition.defer_view_truncation(from_op, Vec::new(), None); - return; - } - reconcile_partition_view_divergence(self.id, partition, Some(&pending)).await; - if partition.consensus().view_log_unproven() { - return; - } - } else { - partition.consensus().settle_view_log_proof(); + partition.defer_view_truncation(from_op..=u64::MAX, Vec::new(), None); + return; + } + reconcile_partition_view_divergence(self.id, partition, Some(&pending)).await; + if partition.view_truncation_pending() + || (partition.consensus().view_log_unproven() + && partition.consensus().view_log_repair_op().is_none()) + { + return; + } + if let Some(missing_op) = partition.consensus().view_log_repair_op() { + self.request_partition_view_repair(partition, missing_op, pending.op_head, None) + .await; + return; } let consensus = partition.consensus(); // Identity, not presence: see the metadata twin. The floor is the local @@ -6199,6 +6240,9 @@ where where B: MessageBus, P: Pipeline<Entry = consensus::PipelineEntry>, + MJ: JournalHandle, + <MJ as JournalHandle>::Target: + Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, { // `ViewChange` too: a parked view change repairs toward its merged log // and cannot start until the window fills. Gating on `Normal` alone @@ -6246,8 +6290,9 @@ where }) }; if let Some((peer, nonce, session_from_op, to_op)) = stalled { - let from_op = - stalled_repair_from_op(session_from_op, consensus.commit_min(), repairing_view); + let from_op = consensus.view_log_repair_op().unwrap_or_else(|| { + stalled_repair_from_op(session_from_op, consensus.commit_min(), repairing_view) + }); if from_op > to_op { // `from_op` past `to_op` without `commit_min` reaching it: the // primary-elect window above starts at the merged log's commit @@ -6318,6 +6363,9 @@ where ) where B: MessageBus, P: Pipeline<Entry = consensus::PipelineEntry>, + MJ: JournalHandle, + <MJ as JournalHandle>::Target: + Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, { *self.metadata_repair.borrow_mut() = None; @@ -6377,36 +6425,8 @@ where .await; } - /// Compare this replica's log against the headers the view decided, and drop or - /// report where they disagree. - /// - /// Without this, divergence is silent and permanent: the replica acks with its - /// own checksum, the primary rejects the ack, and journal repair skips an op - /// it already has a header for. - /// - /// Both roles. A backup runs it against the `StartView` suffix it adopted, the - /// primary-elect against the merged log before the coverage scan in - /// [`Self::advance_pending_metadata_view`]. The merge does NOT reconcile the - /// primary's log for it: nothing installs the merged headers into the journal, - /// `RebuildPipeline` reads the pipeline back out of it, and `CommitJournal` - /// applies whatever sits at each op up to the merged commit point. - /// - /// Runs on every adoption, parked suffix or not: an EMPTY `StartView` suffix - /// (`commit == op`, the steady case) parks no pending log, yet adoption still - /// drops the head under any journaled relics above it, and the primary's next - /// prepare would collide with them in `append` and poison the journal. With - /// no pending log the divergence scan has nothing to walk and only the - /// above-head sweep applies, with the head read off the adopted sequencer. - /// - /// The split at the announced commit point is what matters. Above it a - /// disagreement is ordinary, so the entry is dropped and the primary's - /// retransmission refills the range. At or below it, this replica applied - /// something the view says was different, which only state transfer fixes, so - /// it is reported and left alone. - /// - /// Truncation uses `Journal::truncate_from`, not `drain`: `drain` advances - /// `snapshot_op` past what it removed, marking ops that must stay refillable - /// as evictable. + // The shard pump serializes these rewrites with append. The journal's drain + // guard excludes concurrent compaction, but does not exclude append itself. #[allow(clippy::future_not_send)] async fn reconcile_metadata_view_divergence(&self) where @@ -6414,126 +6434,62 @@ where MJ: JournalHandle, <MJ as JournalHandle>::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, - M: MetadataStm, { let metadata = self.plane.metadata(); - let Some(ref consensus) = metadata.consensus else { - return; - }; - let pending = consensus.pending_view_log(); - let Some(journal) = metadata.journal.as_ref() else { + let (Some(consensus), Some(journal)) = (&metadata.consensus, &metadata.journal) else { return; }; if !consensus.view_log_unproven() { return; } - - // Truncation is safe only above what this replica has *applied*, which is - // not the view's commit point: `pending.commit_max` is the new primary's - // number and a backup can sit above it. Splitting on the view's number - // would drop already-executed ops with no rollback, and silently. - let announced_commit = pending.as_ref().map_or(0, |pending| pending.commit_max); - let applied_floor = consensus.commit_min(); - - let mut repairable_from: Option<u64> = None; - for canonical in pending.as_ref().map_or(&[][..], |pending| &pending.headers) { - let Some(local) = usize::try_from(canonical.op) - .ok() - .and_then(|slot| journal.handle().header(slot)) - else { - continue; - }; - if header_is_view_entry(&local, canonical) { - continue; - } - if canonical.op <= applied_floor { - tracing::error!( - shard = self.id, - op = canonical.op, - view = consensus.view(), - commit_max = announced_commit, - commit_min = consensus.commit_min(), - local_checksum = local.checksum, - canonical_checksum = canonical.checksum, - "committed op {} disagrees with the view that just started; this replica \ - applied a different op as committed and cannot be reconciled by log repair", - canonical.op - ); - continue; - } - repairable_from = Some(repairable_from.map_or(canonical.op, |op| op.min(canonical.op))); - } - - // A suffix ABOVE the announced head is named by nobody, so the loop cannot - // see it, and it is exactly the log that AGREES in-window (restart and - // re-adopt), where `repairable_from` never arms. Adoption drops the head - // under it, and the next prepare at `op_head + 1` then collides in `append`, - // which refuses the slot even when the ops match. Floored at the applied - // point too: an executed op is not rollback-able whatever the head says. - // With no parked suffix the adopted sequencer IS the announced head. + let pending = consensus.pending_view_log(); let op_head = pending.as_ref().map_or_else( || consensus.sequencer().current_sequence(), |pending| pending.op_head, ); - let above_head = op_head.max(applied_floor) + 1; - if journal - .handle() - .last_op() - .is_some_and(|last_op| last_op >= above_head) + let journal = journal.handle(); + let above_head = op_head.max(consensus.commit_min()).saturating_add(1); + if consensus.view_log_repair_op().is_none() + && journal.last_op().is_some_and(|last| last >= above_head) { - repairable_from = Some(repairable_from.map_or(above_head, |op| op.min(above_head))); - } - if let Some(from_op) = consensus::first_unproven_op( - applied_floor, - pending.as_ref().map_or(&[][..], |pending| &pending.headers), - op_head, - journal.handle().last_op(), - |op| { - usize::try_from(op) - .ok() - .and_then(|slot| journal.handle().header(slot).map(|header| *header)) - }, - ) { - repairable_from = Some(repairable_from.map_or(from_op, |op| op.min(from_op))); - } - - let Some(from_op) = repairable_from else { - consensus.settle_view_log_proof(); - return; - }; - // SERIALIZATION: the drain guard excludes `truncate_from` against `drain`, - // NOT against an append; that is metadata's private `journal_gate`. It holds - // by call-site placement, not construction: the pump is single-threaded, and - // both callers run with a view change parked, so no submit is admitted and no - // repair prepare is in flight for these ops. Routing shard-side journal - // mutations through gate-taking metadata methods would make it structural. - match journal.handle().truncate_from(from_op).await { - Ok(removed) => { - consensus.settle_view_log_proof(); - // The snapshot's head and commit point do not move when entries - // are removed under them, so without this the next `DoViewChange` - // would advertise the dropped headers and offer bodies this replica - // cannot serve. - consensus.note_journal_mutation(); - tracing::warn!( - shard = self.id, - from_op, - removed, - op_head, - view = consensus.view(), - "dropped {removed} unproven entries from op {from_op} that disagreed with \ - the view's log; the primary's retransmission refills the range" - ); + if let Err(error) = journal.truncate_from(above_head).await { + tracing::error!(shard = self.id, above_head, %error, "could not truncate superseded metadata suffix"); + return; } - Err(error) => { - tracing::error!( - shard = self.id, - from_op, - %error, - "could not drop the diverging uncommitted entries from op {from_op}; journal \ - repair skips ops it already holds a header for, so this replica will not \ - converge at those ops until it is restarted" - ); + consensus.note_journal_mutation(); + } + loop { + match consensus::view_log_proof( + consensus.commit_min(), + pending.as_ref().map_or(&[][..], |pending| &pending.headers), + op_head, + |op| { + usize::try_from(op) + .ok() + .and_then(|slot| journal.header(slot).map(|header| *header)) + }, + ) { + consensus::ViewLogProof::Proven => { + consensus.settle_view_log_proof(); + return; + } + consensus::ViewLogProof::Missing { + op, + unproven_through, + } => { + consensus.fence_view_log(op, unproven_through); + return; + } + consensus::ViewLogProof::Divergent(op) => { + consensus.fence_view_log(op, op); + // The canonical tail may be this replica's only copy. Remove + // the proven mismatch without cutting entries above a gap. + if let Err(error) = journal.remove_range(op..=op).await { + tracing::error!(shard = self.id, op, %error, "could not remove divergent metadata prepare"); + return; + } + consensus.note_journal_mutation(); + } } } } @@ -6572,7 +6528,7 @@ where // ingest skips an op it already holds a header for, so a diverging entry would // never be replaced. No-op when nothing diverges. self.reconcile_metadata_view_divergence().await; - if consensus.view_log_unproven() { + if consensus.view_log_unproven() && consensus.view_log_repair_op().is_none() { return; } let Some(pending) = consensus.pending_view_log() else { @@ -6591,11 +6547,13 @@ where repair_floor, commit_min: consensus.commit_min(), }; - let missing = first_op_not_covered(&pending, floor, |op| { - usize::try_from(op) - .ok() - .and_then(|slot| journal.handle().header(slot)) - .map(|header| *header) + let missing = consensus.view_log_repair_op().or_else(|| { + first_op_not_covered(&pending, floor, |op| { + usize::try_from(op) + .ok() + .and_then(|slot| journal.handle().header(slot)) + .map(|header| *header) + }) }); let Some(missing_op) = missing else { @@ -6732,7 +6690,19 @@ where where B: MessageBus, P: Pipeline<Entry = consensus::PipelineEntry>, + MJ: JournalHandle, + <MJ as JournalHandle>::Target: + Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, { + if !consensus.is_transferring() + && consensus.view_log_unproven() + && consensus.view_log_repair_op().is_none() + { + self.reconcile_metadata_view_divergence().await; + if consensus.view_log_unproven() && consensus.view_log_repair_op().is_none() { + return; + } + } // Never against self. A self-addressed `RequestPrepares` cannot be // delivered (the replica registry holds no entry for this node), and the // send fails AFTER the session is recorded, so the session would stand @@ -6744,11 +6714,17 @@ where } if consensus.is_normal() && !consensus.is_transferring() - && consensus.commit_min() < consensus.commit_max() + && (consensus.commit_min() < consensus.commit_max() || consensus.view_log_unproven()) && self.metadata_repair.borrow().is_none() { - let to_op = consensus.commit_max(); - let from_op = consensus.commit_min() + 1; + let to_op = if consensus.view_log_unproven() { + consensus.sequencer().current_sequence() + } else { + consensus.commit_max() + }; + let from_op = consensus + .view_log_repair_op() + .unwrap_or_else(|| consensus.commit_min() + 1); // Spent here rather than at the detector, so the edge-triggered // sites spend it too: an edge-armed repair that completes before // the next tick would otherwise leave the count saturated and hand @@ -8085,6 +8061,10 @@ where ); continue; } + let repair_from = partition + .consensus() + .view_log_repair_op() + .unwrap_or_else(|| commit_min.saturating_add(1)); let due = partition.repair.as_mut().and_then(|session| { if !session_live { return None; @@ -8097,7 +8077,7 @@ where Some(( session.peer, session.nonce, - commit_min.saturating_add(1), + repair_from, session.fetch_to_op, cluster, self_id, @@ -8576,7 +8556,7 @@ where if !self.partition_incarnation_matches( partition, header.partition_incarnation, - "RequestStateTransfer", + Command::RequestStateTransfer, ) { return; } @@ -8782,7 +8762,7 @@ where if !self.partition_incarnation_matches( partition, header.partition_incarnation, - "RequestStateChunk", + Command::RequestStateChunk, ) { return; } @@ -9216,10 +9196,32 @@ where where B: MessageBus, { - let consensus = partition.consensus(); - if !consensus.is_normal() || consensus.is_transferring() || partition.repair.is_some() { + if !partition.consensus().is_normal() + || partition.consensus().is_transferring() + || partition.repair.is_some() + || partition.view_truncation_pending() + { return false; } + if partition.consensus().view_log_unproven() + && partition.consensus().view_log_repair_op().is_none() + { + let pending = partition.consensus().pending_view_log(); + if partition.has_io_dispatcher() + && let Some(from_op) = partition_view_divergence_from(partition, pending.as_ref()) + { + partition.defer_view_truncation(from_op..=u64::MAX, Vec::new(), Some(peer)); + return false; + } + reconcile_partition_view_divergence(self.id, partition, pending.as_ref()).await; + if partition.view_truncation_pending() + || (partition.consensus().view_log_unproven() + && partition.consensus().view_log_repair_op().is_none()) + { + return false; + } + } + let consensus = partition.consensus(); // Read, never scanned: the tally is republished by each sweep (see // `tick_partitions`), so it is at worst one tick stale, which is all a // concurrency ceiling needs. Callers here hold a `&mut` to one @@ -9254,31 +9256,32 @@ where let commit_to_op = consensus.commit_max(); let commit_lag = consensus.commit_min() < commit_to_op; let head = consensus.sequencer().current_sequence(); - if !commit_lag && head <= commit_to_op { + if !commit_lag && head <= commit_to_op && !consensus.view_log_unproven() { return false; } let missing_suffix = partition_missing_suffix_through(partition); // Fetch the adopted suffix even while committed operations lag. Later // live prepares can advance the head while an adopted body is missing. - let Some(fetch_to_op) = + let Some(fetch_to_op) = consensus.view_log_repair_op().map(|_| head).or_else(|| { partition_repair_fetch_to_op(consensus.commit_min(), commit_to_op, missing_suffix) - else { + }) else { return false; }; // A resident window may only be waiting for persistence or a bounded commit walk. - if partition - .log - .journal() - .inner - .repaired_window_shape(consensus.commit_min(), fetch_to_op) - .complete + if !consensus.view_log_unproven() + && partition + .log + .journal() + .inner + .repaired_window_shape(consensus.commit_min(), fetch_to_op) + .complete { return false; } let nonce = iggy_common::random_id::get_uuid(); - let from_op = consensus.commit_min() + 1; - let cluster = consensus.cluster(); - let self_id = consensus.replica(); + let from_op = consensus + .view_log_repair_op() + .unwrap_or_else(|| consensus.commit_min() + 1); let namespace = consensus.group(); // Spent here for the same reason the ceiling is: the four edge-triggered // sites never touch it, so a short edge-armed repair would leave the @@ -9305,8 +9308,8 @@ where "partition behind the group frontier; requesting repair" ); self.send_request_prepares( - cluster, - self_id, + partition.consensus().cluster(), + partition.consensus().replica(), peer, nonce, from_op, @@ -9429,7 +9432,7 @@ where if !self.partition_incarnation_matches( partition, header.partition_incarnation, - "StateTransferTarget", + Command::StateTransferTarget, ) { return; } @@ -9657,7 +9660,7 @@ where if !self.partition_incarnation_matches( partition, header.partition_incarnation, - "StateChunk", + Command::StateChunk, ) { return; } @@ -10741,6 +10744,7 @@ where commit, op, pending.as_ref().map(|pending| pending.headers.as_slice()), + consensus.view_log_unproven_range(), )); } @@ -10802,6 +10806,7 @@ where op, |entry_op| window.get(&entry_op).copied(), pending.as_ref().map(|pending| pending.headers.as_slice()), + consensus.view_log_unproven_range(), ); consensus.set_local_dvc_suffix(suffix); } @@ -11448,6 +11453,7 @@ fn build_metadata_dvc_suffix<J>( commit: u64, op: u64, view_headers: Option<&[PrepareHeader]>, + unproven: std::ops::RangeInclusive<u64>, ) -> DvcSuffix where J: JournalHandle, @@ -11467,6 +11473,7 @@ where .map(|header| *header) }, view_headers, + unproven, ) } @@ -11477,6 +11484,7 @@ fn build_dvc_suffix( op: u64, header_at: impl Fn(u64) -> Option<PrepareHeader>, view_headers: Option<&[PrepareHeader]>, + unproven: std::ops::RangeInclusive<u64>, ) -> DvcSuffix { // Stitch the adopted view's headers over the journal, high-to-low. // @@ -11528,7 +11536,8 @@ fn build_dvc_suffix( let mut nack_bitset = 0u128; let mut present_bitset = 0u128; for (index, entry_op) in (low..=op).rev().enumerate() { - if let Some(header) = header_at(entry_op) { + let resident = header_at(entry_op); + if let Some(header) = resident.filter(|_| !unproven.contains(&entry_op)) { headers.push(header); // A header in the index means the entry is in the WAL at a known // offset, the same condition `on_request_prepares` serves from. @@ -11543,9 +11552,9 @@ fn build_dvc_suffix( headers.push(*header); } else { headers.push(dvc_blank(entry_op)); - if entry_op > commit { + if entry_op > commit && resident.is_none() { nack_bitset |= 1u128 << index; - } else { + } else if entry_op <= commit { // The commit point, the one slot that goes out blank AND // un-nacked. The merge scans it and may not discard it, so a // sender is asking the new primary to take the header from @@ -11570,16 +11579,8 @@ fn build_dvc_suffix( DvcSuffix::new(headers, nack_bitset, present_bitset) } -/// Partition-plane twin of `Shard::reconcile_metadata_view_divergence`, preserving -/// applied and flushed history while dropping an unproven local suffix. -/// -/// Worse to skip here than on the metadata plane, which is why this exists. -/// Partition `append` has no slot-collision check, so a re-prepared op pushes a -/// duplicate header and rewrites `op_to_storage_offset`. Materialization walks -/// positionally, so it would flush the stale entry to the segment: -/// durable divergent bytes, no error anywhere. +/// Only the suffix above the adopted head is safe to truncate as a whole. fn partition_view_divergence_from<B, SB>( - shard: u16, partition: &IggyPartition<B, SB>, pending: Option<&MergedLog>, ) -> Option<u64> @@ -11587,72 +11588,27 @@ where B: MessageBus, SB: SuperblockStore, { - if !partition.consensus().view_log_unproven() { + if !partition.consensus().view_log_unproven() + || partition.consensus().view_log_repair_op().is_some() + { return None; } - // A flushed entry is already committed even if its apply walk is pending. - // The announced commit point does not prove the identity of local entries. - let announced_commit = pending.map_or(0, |pending| pending.commit_max); let applied_floor = partition .consensus() .commit_min() .max(partition.log.journal().inner.evicted_through()); - - let mut repairable_from: Option<u64> = None; - for canonical in pending.map_or(&[][..], |pending| &pending.headers) { - let Some(local) = partition.log.journal().inner.header_by_op(canonical.op) else { - continue; - }; - if header_is_view_entry(&local, canonical) { - continue; - } - if canonical.op <= applied_floor { - tracing::error!( - shard, - namespace_raw = partition.consensus().group(), - op = canonical.op, - view = partition.consensus().view(), - commit_max = announced_commit, - commit_min = partition.consensus().commit_min(), - local_checksum = local.checksum, - canonical_checksum = canonical.checksum, - "committed partition op {} disagrees with the view that just started; this \ - replica applied a different op and log repair cannot reconcile it", - canonical.op - ); - continue; - } - repairable_from = Some(repairable_from.map_or(canonical.op, |op| op.min(canonical.op))); - } - - // The suffix above the announced head, which no canonical header names. As on - // the metadata twin, except here `append` pushes a duplicate rather than - // erroring. With no parked suffix the adopted sequencer IS the announced head. let op_head = pending.map_or_else( || partition.consensus().sequencer().current_sequence(), |pending| pending.op_head, ); - let above_head = op_head.max(applied_floor) + 1; - if partition + let above_head = op_head.max(applied_floor).checked_add(1)?; + partition .log .journal() .inner .last_op() - .is_some_and(|last_op| last_op >= above_head) - { - repairable_from = Some(repairable_from.map_or(above_head, |op| op.min(above_head))); - } - - if let Some(from_op) = consensus::first_unproven_op( - applied_floor, - pending.map_or(&[][..], |pending| &pending.headers), - op_head, - partition.log.journal().inner.last_op(), - |op| partition.log.journal().inner.header_by_op(op), - ) { - repairable_from = Some(repairable_from.map_or(from_op, |op| op.min(from_op))); - } - repairable_from + .filter(|last| *last >= above_head) + .map(|_| above_head) } #[allow(clippy::future_not_send)] @@ -11664,34 +11620,55 @@ async fn reconcile_partition_view_divergence<B, SB>( B: MessageBus, SB: journal::superblock::SuperblockStore, { - let repairable_from = partition_view_divergence_from(shard, partition, pending); - let Some(from_op) = repairable_from else { - partition.consensus().settle_view_log_proof(); + if !partition.consensus().view_log_unproven() || partition.view_truncation_pending() { return; - }; - match partition.truncate_uncommitted_from(from_op).await { - Ok(removed) => { - partition.consensus().settle_view_log_proof(); - tracing::warn!( - shard, - namespace_raw = partition.consensus().group(), - from_op, - removed, - view = partition.consensus().view(), - "dropped {removed} unproven partition entries from op {from_op} that \ - disagreed with the view's log; the primary's retransmission refills the range" - ); - } - Err(error) => { - tracing::error!( - shard, - namespace_raw = partition.consensus().group(), - from_op, - %error, - "could not drop the diverging uncommitted partition entries from op \ - {from_op}; repair skips ops it already holds, so this replica will not \ - converge there until restarted" - ); + } + if let Some(from_op) = partition_view_divergence_from(partition, pending) + && let Err(error) = partition.truncate_uncommitted_from(from_op).await + { + tracing::error!(shard, from_op, %error, "could not truncate superseded partition suffix"); + return; + } + let op_head = pending.map_or_else( + || partition.consensus().sequencer().current_sequence(), + |pending| pending.op_head, + ); + loop { + let floor = partition + .consensus() + .commit_min() + .max(partition.log.journal().inner.evicted_through()); + match consensus::view_log_proof( + floor, + pending.map_or(&[][..], |pending| &pending.headers), + op_head, + |op| partition.log.journal().inner.header_by_op(op), + ) { + consensus::ViewLogProof::Proven => { + partition.consensus().settle_view_log_proof(); + return; + } + consensus::ViewLogProof::Missing { + op, + unproven_through, + } => { + partition.consensus().fence_view_log(op, unproven_through); + return; + } + consensus::ViewLogProof::Divergent(op) => { + partition.consensus().fence_view_log(op, op); + if partition.has_io_dispatcher() { + let consensus = partition.consensus(); + let peer = (!consensus.is_primary()) + .then(|| consensus.primary_index(consensus.view())); + partition.defer_view_truncation(op..=op, Vec::new(), peer); + return; + } + if let Err(error) = partition.remove_divergent_prepare(op).await { + tracing::error!(shard, op, %error, "could not remove divergent partition prepare"); + return; + } + } } } } @@ -11989,10 +11966,16 @@ async fn dispatch_vsr_actions<B, P, J>( target, group, } => { + if consensus.view_log_unproven() { + continue; + } let Some(journal) = journal else { continue; }; for op in *from_op..=*to_op { + if !consensus.view_log_entry_proven(op) { + continue; + } let Some(prepare_header) = journal.handle().header(op as usize) else { continue; }; @@ -12670,7 +12653,7 @@ mod dvc_suffix_window_tests { // Reading the journal alone reports them blank AND nacked, which reaches a // nack quorum against ops the view had just decided to keep. let view = view_headers(3, 5); - let suffix = build_dvc_suffix(2, 0, |_| None, Some(&view)); + let suffix = build_dvc_suffix(2, 0, |_| None, Some(&view), 0..=0); assert_eq!( suffix.len(), @@ -12697,7 +12680,7 @@ mod dvc_suffix_window_tests { #[test] fn given_no_adopted_view_when_the_journal_is_empty_should_nack() { // The contrast: without an adopted view the same holes really are proof. - let suffix = build_dvc_suffix(2, 5, |_| None, None); + let suffix = build_dvc_suffix(2, 5, |_| None, None, 0..=0); assert_eq!( suffix.nack_bitset(), 0b0111, @@ -12705,12 +12688,30 @@ mod dvc_suffix_window_tests { ); } + #[test] + fn given_an_unproven_prefix_when_building_dvc_should_neither_advertise_nor_nack_held_entries() { + let view = view_headers(5, 6); + let suffix = build_dvc_suffix(2, 6, |op| Some(held(op)), Some(&view), 3..=4); + assert_eq!( + suffix.present_bitset(), + 0b10011, + "only ops 2, 5 and 6 are proven" + ); + assert_eq!( + suffix.nack_bitset(), + 0, + "unknown held entries cannot be nacked" + ); + assert_eq!(suffix.headers()[2], consensus::dvc_blank(4)); + assert_eq!(suffix.headers()[3], consensus::dvc_blank(3)); + } + #[test] fn given_an_adopted_view_when_the_journal_covers_part_should_prefer_the_journal() { // Journal first, so an op whose body this replica can serve keeps its // present bit; the view fills only what the journal is missing. let view = view_headers(3, 5); - let suffix = build_dvc_suffix(2, 5, |op| (op == 5).then(|| held(op)), Some(&view)); + let suffix = build_dvc_suffix(2, 5, |op| (op == 5).then(|| held(op)), Some(&view), 0..=0); assert_eq!(suffix.len(), 4); assert_eq!(suffix.present_bitset(), 0b0001, "only op 5 is servable"); @@ -12718,7 +12719,7 @@ mod dvc_suffix_window_tests { // The head is the max of the two, never the view's alone. let short_view = view_headers(3, 4); - let deeper = build_dvc_suffix(2, 6, |op| Some(held(op)), Some(&short_view)); + let deeper = build_dvc_suffix(2, 6, |op| Some(held(op)), Some(&short_view), 0..=0); assert_eq!(deeper.headers().first().map(|header| header.op), Some(6)); assert_eq!(deeper.present_bitset(), 0b1_1111, "ops 6 down to 2"); } @@ -12728,7 +12729,7 @@ mod dvc_suffix_window_tests { // A blank is the view saying "no header here", not one this replica holds. let mut view = view_headers(3, 5); view[1] = consensus::dvc_blank(4); - let suffix = build_dvc_suffix(2, 0, |_| None, Some(&view)); + let suffix = build_dvc_suffix(2, 0, |_| None, Some(&view), 0..=0); assert_eq!(suffix.nack_bitset(), 0b010, "only the blank op nacks"); } @@ -12740,7 +12741,7 @@ mod dvc_suffix_window_tests { // that op and may not discard it, so a quorum of these deadlocks the view // change. Pinned here because both compaction paths are meant to keep the // header alive precisely so this shape never leaves a healthy replica. - let suffix = build_dvc_suffix(5, 5, |_| None, None); + let suffix = build_dvc_suffix(5, 5, |_| None, None, 0..=0); assert_eq!(suffix.len(), 1); assert_eq!(floor(&suffix), Some(5)); @@ -12760,7 +12761,7 @@ mod dvc_suffix_window_tests { let depth = DVC_HEADERS_MAX as u64 - 1; let commit = 500; let op = commit + depth; - let suffix = build_dvc_suffix(commit, op, |op| Some(held(op)), None); + let suffix = build_dvc_suffix(commit, op, |op| Some(held(op)), None, 0..=0); assert_eq!(suffix.len(), DVC_HEADERS_MAX, "the widest window that fits"); assert_eq!( @@ -12776,7 +12777,7 @@ mod dvc_suffix_window_tests { // sender's commit, with no marker on the frame saying so. let commit = 500; let op = commit + DVC_HEADERS_MAX as u64; - let suffix = build_dvc_suffix(commit, op, |op| Some(held(op)), None); + let suffix = build_dvc_suffix(commit, op, |op| Some(held(op)), None, 0..=0); assert_eq!(suffix.len(), DVC_HEADERS_MAX); assert_eq!( @@ -12787,7 +12788,7 @@ mod dvc_suffix_window_tests { assert!(floor(&suffix) > Some(commit)); // Second path, at any depth: ops are 1-based, so commit 0 floors at op 1. - let from_zero = build_dvc_suffix(0, 3, |op| Some(held(op)), None); + let from_zero = build_dvc_suffix(0, 3, |op| Some(held(op)), None, 0..=0); assert_eq!(floor(&from_zero), Some(1)); } @@ -12800,7 +12801,7 @@ mod dvc_suffix_window_tests { // the evicted ring); nothing in `build_dvc_suffix` enforces it. let commit = 500; let compacted = |op: u64| (op >= commit).then(|| held(op)); - let suffix = build_dvc_suffix(commit, commit + 3, compacted, None); + let suffix = build_dvc_suffix(commit, commit + 3, compacted, None, 0..=0); let commit_index = suffix.index_of(commit + 3, commit).expect("in window"); assert!( @@ -13785,6 +13786,7 @@ mod partition_ack_durability_tests { backup.sequencer().current_sequence(), |_| None, None, + 0..=0, ); assert_eq!( suffix.nack_bitset(), diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs index fafd3e5b9..21fcf5931 100644 --- a/core/shard/src/partition_io.rs +++ b/core/shard/src/partition_io.rs @@ -705,10 +705,16 @@ where } } partitions::PartitionIoStep::ViewApplied { actions, peer } => { - partition.consensus().settle_view_log_proof(); + let pending = partition.consensus().pending_view_log(); + crate::reconcile_partition_view_divergence(self.id, partition, pending.as_ref()) + .await; if let Some(peer) = peer { - self.finish_partition_view_adoption(namespace, peer, actions) - .await; + if partition.consensus().view_log_unproven() { + self.maybe_request_partition_repair(partition, peer).await; + } else { + self.finish_partition_view_adoption(namespace, peer, actions) + .await; + } } else { self.advance_pending_partition_view(namespace).await; } @@ -803,6 +809,7 @@ mod tests { ConsumerGroupOffsets, ConsumerKind, ConsumerOffsets, Durability, IggyByteSize, IggyTimestamp, PartitionStats, TopicRuntimeOptions, variadic, }; + use journal::Journal; use journal::prepare_journal::PrepareJournal; use journal::superblock::{SuperblockContents, SuperblockStore}; use message_bus::{IggyMessageBus, MessageBus}; @@ -4160,6 +4167,80 @@ mod tests { type IoTestShard<B = Rc<IggyMessageBus>> = IggyShard<B, PrepareJournal, (), IoTestMetadata, PapayaShardsTable, HeldSuperblock>; + #[compio::test] + async fn given_a_stale_primary_heartbeat_when_the_suffix_cache_is_stale_should_reply_with_fresh_headers() + { + const HEAD: u64 = 2; + const INCARNATION: u64 = 1; + let bus = Rc::new(IggyMessageBus::new(0)); + let sent = Rc::new(RefCell::new(Vec::new())); + let captured = Rc::clone(&sent); + bus.set_replica_forward_fn(Box::new(move |_, _, frame| { + captured.borrow_mut().push(frame); + Ok(()) + })); + for replica in [0, 2] { + assert!(bus.owner_table().try_claim(replica, 1)); + } + let (mut owner, _sender) = test_owner(&bus, None); + owner.partition_consensus.replica_count = 3; + let namespace = IggyNamespace::new(0, 0, 0); + let prepares = primary_prepares(&bus, namespace, HEAD).await; + let mut partition = repairing_replica(&bus, namespace, (REPAIR_PRIMARY, 3), 0, 0, 0); + partition.repair = None; + partition.set_created_revision(INCARNATION); + crate::refresh_partition_dvc_suffix(&partition); + for prepare in &prepares { + partition + .log + .journal() + .inner + .append(prepare.clone().into_frozen()) + .await + .unwrap(); + partition.consensus().note_journal_mutation(); + } + partition.consensus().sequencer().set_sequence(HEAD); + assert!(partition.consensus().local_dvc_suffix_stale()); + owner.plane.partitions().remove(&namespace).unwrap(); + owner.plane.partitions().insert(namespace, partition); + sent.borrow_mut().clear(); + let heartbeat = Message::<CommitHeader>::new(size_of::<CommitHeader>()).transmute_header( + |_, header: &mut CommitHeader| { + header.command = Command::Commit; + header.cluster = 1; + header.group = namespace.inner(); + header.partition_incarnation = INCARNATION; + header.replica = 0; + header.view = REPAIR_VIEW - 1; + header.size = u32::try_from(size_of::<CommitHeader>()).unwrap(); + header.seal(); + }, + ); + + owner.on_commit(&heartbeat).await; + + assert_eq!(sent.borrow().len(), 2); + let expected: Vec<_> = prepares + .iter() + .rev() + .map(|prepare| *prepare.header()) + .collect(); + for frame in sent.borrow().iter() { + let reply = + Message::<StartViewHeader>::try_from(Owned::copy_from_slice(frame.as_slice())) + .unwrap(); + let suffix = consensus::dvc_suffix_decode( + &reply.as_slice()[size_of::<StartViewHeader>()..], + HEAD, + 0, + 0, + ) + .unwrap(); + assert_eq!(suffix.headers(), expected); + } + } + #[compio::test] async fn given_unproven_suffix_when_adopting_start_view_should_repair_before_applying() { const APPLIED: u64 = 2; @@ -4195,21 +4276,138 @@ mod tests { .await; let partition = owner.plane.partitions().get_by_ns(&namespace).unwrap(); assert_eq!(partition.consensus().commit_min(), APPLIED); - assert!(!partition.log.journal().inner.holds_op(APPLIED + 1)); + assert!(partition.log.journal().inner.holds_op(APPLIED + 1)); + assert!(partition.consensus().view_log_unproven()); + assert!(!partition.consensus().view_log_entry_proven(APPLIED + 1)); + for prepare in prepares.iter().rev() { + let repaired = prepare.clone().transmute_header( + |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { + repaired.0 = header; + repaired.0.command = Command::RepairPrepare; + }, + ); + owner.on_repair_prepare(repaired).await; + } + let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); assert!(!partition.consensus().view_log_unproven()); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + assert_eq!(partition.consensus().commit_min(), NEW_HEAD); + } + + #[compio::test] + async fn given_a_divergent_entry_when_removal_is_deferred_should_keep_the_canonical_tail() { + const APPLIED: u64 = 2; + const DIVERGENT: u64 = 3; + const HEAD: u64 = 4; + const NEW_VIEW: u32 = 3; + let bus = Rc::new(IggyMessageBus::new(0)); + let (owner, _sender) = test_owner(&bus, None); + let namespace = IggyNamespace::new(0, 0, 0); + let prepares = primary_prepares(&bus, namespace, HEAD).await; + let mut partition = + repairing_replica(&bus, namespace, (REPAIR_BACKUP, 3), REPAIR_PRIMARY, 0, 0); + partition.repair = None; + for prepare in &prepares { + partition.on_replicate(prepare.clone()).await; + } + partition.consensus().advance_commit_max(APPLIED); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + partition + .log + .journal() + .inner + .remove_range(DIVERGENT..=DIVERGENT) + .await + .unwrap(); + let divergent = prepares[usize::try_from(DIVERGENT - 1).unwrap()] + .clone() + .transmute_header(|previous, header: &mut PrepareHeader| { + *header = previous; + header.client += 1; + header.checksum = header.identity_checksum(); + }); + partition + .log + .journal() + .inner + .append(divergent.into_frozen()) + .await + .unwrap(); + let partitions = owner.plane.partitions(); + partitions.remove(&namespace).unwrap(); + partitions.insert(namespace, partition); + partitions.set_io_notifier( + owner.partition_io.notifier(), + owner.partition_io.limits.bytes_max(), + ); + + owner + .on_start_view(start_view_suffix( + namespace, + NEW_VIEW, + HEAD, + HEAD, + &[*prepares.last().unwrap().header()], + )) + .await; + assert!( + partitions + .get_by_ns(&namespace) + .unwrap() + .view_truncation_pending() + ); + for _ in 0..crate::router::COOPERATIVE_EVENT_BUDGET { + owner.service_partition_io().await; + if !partitions + .get_by_ns(&namespace) + .unwrap() + .view_truncation_pending() + { + break; + } + } + let partition = partitions.get_by_ns(&namespace).unwrap(); + assert!(!partition.view_truncation_pending()); + assert!(partition.consensus().view_log_unproven()); + assert_eq!(partition.consensus().commit_min(), APPLIED); + assert!(!partition.log.journal().inner.holds_op(DIVERGENT)); + assert_eq!( + partition.log.journal().inner.header_by_op(HEAD), + Some(*prepares.last().unwrap().header()) + ); + + let repaired = prepares[usize::try_from(DIVERGENT - 1).unwrap()] + .clone() + .transmute_header( + |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { + repaired.0 = header; + repaired.0.command = Command::RepairPrepare; + }, + ); + owner.on_repair_prepare(repaired).await; + let partition = partitions.get_mut_by_ns(&namespace).unwrap(); + assert!(!partition.consensus().view_log_unproven()); + partition.commit_journal(partitions.config()).await; + assert_eq!(partition.consensus().commit_min(), HEAD); } #[compio::test] async fn given_deferred_view_truncation_when_resuming_should_settle_proof_only_after_success() { const APPLIED: u64 = 2; - const LOCAL_HEAD: u64 = 4; - const NEW_HEAD: u64 = 6; + const LOCAL_HEAD: u64 = 6; + const NEW_HEAD: u64 = 4; const NEW_VIEW: u32 = 3; - for truncate_fails in [false, true] { + for (truncate_fails, repair_fallback) in + [(false, false), (true, false), (false, true), (true, true)] + { let bus = Rc::new(IggyMessageBus::new(0)); let (owner, _sender) = test_owner(&bus, None); let namespace = IggyNamespace::new(0, 0, 0); - let prepares = primary_prepares(&bus, namespace, NEW_HEAD).await; + let prepares = primary_prepares(&bus, namespace, LOCAL_HEAD).await; let mut partition = repairing_replica(&bus, namespace, (REPAIR_BACKUP, 3), REPAIR_PRIMARY, 0, 0); partition.repair = None; @@ -4227,15 +4425,28 @@ mod tests { owner.partition_io.notifier(), owner.partition_io.limits.bytes_max(), ); - owner - .on_start_view(start_view_suffix( - namespace, - NEW_VIEW, - NEW_HEAD, - NEW_HEAD, - &[*prepares.last().unwrap().header()], - )) - .await; + let view = start_view_suffix( + namespace, + NEW_VIEW, + NEW_HEAD, + NEW_HEAD, + &[*prepares[usize::try_from(NEW_HEAD - 1).unwrap()].header()], + ); + if repair_fallback { + let partition = partitions.get_mut_by_ns(&namespace).unwrap(); + partition.consensus().handle_start_view( + consensus::PlaneKind::Partitions, + view.header(), + view.body(), + ); + assert!( + !owner + .maybe_request_partition_repair(partition, REPAIR_PRIMARY) + .await + ); + } else { + owner.on_start_view(view).await; + } let partition = partitions.get_mut_by_ns(&namespace).unwrap(); assert!(partition.view_truncation_pending()); assert!(partition.consensus().view_log_unproven()); @@ -4258,9 +4469,12 @@ mod tests { assert!(!partition.view_truncation_pending()); assert_eq!(partition.consensus().view_log_unproven(), truncate_fails); partition.commit_journal(partitions.config()).await; - assert_eq!(partition.consensus().commit_min(), APPLIED); assert_eq!( - partition.log.journal().inner.holds_op(APPLIED + 1), + partition.consensus().commit_min(), + if truncate_fails { APPLIED } else { NEW_HEAD } + ); + assert_eq!( + partition.log.journal().inner.holds_op(NEW_HEAD + 1), truncate_fails ); } @@ -4365,8 +4579,8 @@ mod tests { ); let pending = partition.consensus().pending_view_log(); assert_eq!( - crate::partition_view_divergence_from(0, &partition, pending.as_ref()), - Some(STAMPED) + crate::partition_view_divergence_from(&partition, pending.as_ref()), + None ); assert!( partition @@ -4427,7 +4641,7 @@ mod tests { .partitions() .get_mut_by_ns(&namespace) .unwrap() - .defer_view_truncation(3, Vec::new(), None); + .defer_view_truncation(3..=u64::MAX, Vec::new(), None); owner.on_request_prepares(&request).await; assert!( sent.borrow().is_empty(), diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs index c8cab28e2..1383db4dd 100644 --- a/core/simulator/src/bin/workload-fuzz.rs +++ b/core/simulator/src/bin/workload-fuzz.rs @@ -663,6 +663,13 @@ fn run_quiesce_phase( args.min_owners_compared, convergence.owners_compared, )); } + if sim.is_shell() && args.plane == Plane::Lifecycle && workload.owner_fence_refusals() == 0 { + vacuous(format_args!( + "lifecycle --shell requires an owner-fence refusal, but observed none \ + (stale_owner_stores={}, seed={seed:#x})", + workload.stale_owner_stores(), + )); + } // Again after the drain: the drain both answers outstanding requests and // issues its own resends, so the pre-drain numbers are not the final ones. print_coverage(workload); diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs index 5b3f94307..b9b19cc3e 100644 --- a/core/simulator/src/deps.rs +++ b/core/simulator/src/deps.rs @@ -215,15 +215,15 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal for SimJournal<S> { /// Drop the suffix, so a simulated backup whose entries disagree with a started /// view reconciles the way a real one does. Mirrors /// `PrepareJournal::truncate_from`, whose watermark stays put; here it never moves. - async fn truncate_from(&self, from_op: u64) -> std::io::Result<usize> { + async fn remove_range(&self, ops: std::ops::RangeInclusive<u64>) -> std::io::Result<usize> { #[cfg(test)] if self.fail_next_truncate.replace(false) { return Err(std::io::Error::other("injected journal truncation failure")); } - if from_op == 0 { + if *ops.start() == 0 { return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, - "truncate_from: ops are 1-based, so 0 would discard the whole journal", + "journal removal requires 1-based operations", )); } #[cfg(debug_assertions)] @@ -233,7 +233,7 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal for SimJournal<S> { let doomed: Vec<u64> = headers .keys() .copied() - .filter(|op| *op >= from_op) + .filter(|op| ops.contains(op)) .collect(); for op in &doomed { headers.remove(op); diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 0290f5f05..eb0cab75a 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -3683,15 +3683,16 @@ mod tests { .as_ref() .unwrap() .commit_min(); - assert!(registered > 0); for _ in 0..SETUP_TOTAL_STEPS { - let backup = sim.replicas[BACKUP].shards[0] + if sim.replicas[BACKUP].shards[0] .plane .metadata() .consensus .as_ref() - .unwrap(); - if backup.commit_min() == registered { + .unwrap() + .commit_min() + == registered + { break; } sim.step(); @@ -3704,15 +3705,15 @@ mod tests { let consensus = metadata.consensus.as_ref().unwrap(); let journal = metadata.journal.as_ref().unwrap(); let applied = consensus.commit_min(); + assert_eq!(applied, registered); let source_metadata = primary.plane.metadata(); let source = source_metadata.consensus.as_ref().unwrap(); - assert_eq!(source.sequencer().current_sequence(), applied); let mut prepares = Vec::new(); for name in [ - "old-suffix-first", - "old-suffix-replaced", - "new-suffix-next", - "new-suffix-head", + "retained-first", + "retained-second", + "repair-gap", + "repair-head", ] { let prepare = consensus::seal_prepare_capacity( client.create_stream(name).project(source), @@ -3722,92 +3723,41 @@ mod tests { source.set_last_prepare_checksum(prepare.header().checksum); prepares.push(prepare); } + for (index, prepare) in prepares.iter().take(LOCAL_ENTRIES).enumerate() { + let entry = if fail_truncation && index == 0 { + prepare + .clone() + .transmute_header(|mut header, next: &mut PrepareHeader| { + header.client += 1; + header.checksum = header.identity_checksum(); + *next = header; + }) + } else { + prepare.clone() + }; + journal.handle().append(entry).await.unwrap(); + } let old_head = prepares[LOCAL_ENTRIES - 1].header().op; let suffix_head = *prepares.last().unwrap().header(); - let commit = Message::<iggy_binary_protocol::CommitHeader>::new(size_of::< - iggy_binary_protocol::CommitHeader, - >( - )) - .transmute_header( - |_, header: &mut iggy_binary_protocol::CommitHeader| { - header.command = Command::Commit; - header.cluster = consensus.cluster(); - header.group = METADATA_GROUP; - header.replica = source.replica(); - header.view = source.view(); - header.commit = old_head; - header.timestamp_monotonic = old_head; - header.size = u32::try_from(std::mem::size_of_val(header)).unwrap(); - header.seal(); - }, - ); - shard.on_message(MessageBag::Commit(commit)).await; - assert_eq!( - shard.metadata_repair_window(), - Some((old_head, source.replica())) - ); - let repair_request = outbox - .drain() - .into_iter() - .find_map(|envelope| match envelope.payload { - bus::EnvelopePayload::Replica(message) - if message.header().command == Command::RequestPrepares => - { - Some(*bytemuck::checked::from_bytes::< - iggy_binary_protocol::RequestPreparesHeader, - >( - &message.as_slice() - [..size_of::<iggy_binary_protocol::RequestPreparesHeader>()], - )) - } - _ => None, - }) - .expect("the commit gap arms a real repair session"); - for prepare in prepares.into_iter().take(LOCAL_ENTRIES) { - let repaired = prepare.transmute_header( - |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { - repaired.0 = header; - repaired.0.command = Command::RepairPrepare; - }, - ); - shard.on_message(MessageBag::RepairPrepare(repaired)).await; - } - assert_eq!(consensus.commit_min(), applied); - assert_eq!(consensus.sequencer().current_sequence(), old_head); - let first_unapplied = usize::try_from(applied + 1).unwrap(); - assert!(journal.handle().header(first_unapplied).is_some()); - let peer_request = - Message::<iggy_binary_protocol::RequestPreparesHeader>::new(size_of::< - iggy_binary_protocol::RequestPreparesHeader, - >( - )) - .transmute_header( - |_, header: &mut iggy_binary_protocol::RequestPreparesHeader| { - *header = repair_request; - header.replica = source.replica(); - header.seal(); - }, - ); - shard - .on_message(MessageBag::RequestPrepares(peer_request.clone())) - .await; - assert!( - outbox.drain().into_iter().any(|envelope| { - matches!(envelope.payload, bus::EnvelopePayload::Replica(message) - if message.header().command == Command::RepairPrepare) - }), - "the resident entries must be servable before adoption" - ); + consensus.sequencer().set_sequence(old_head); + let canonical: Vec<_> = if fail_truncation { + prepares + .iter() + .rev() + .map(|prepare| *prepare.header()) + .collect() + } else { + vec![suffix_head] + }; if fail_truncation { journal.fail_next_truncate(); } - let header_size = size_of::<iggy_binary_protocol::StartViewHeader>(); let mut message = Message::<iggy_binary_protocol::StartViewHeader>::new( - header_size + size_of::<PrepareHeader>(), + header_size + std::mem::size_of_val(canonical.as_slice()), ); consensus::encode_prepare_headers( - &[suffix_head], + &canonical, &mut message.as_mut_slice()[header_size..], ); let checksum_body = u128::from(iggy_common::calculate_checksum( @@ -3828,58 +3778,90 @@ mod tests { header.seal(); }, ); - shard.on_message(MessageBag::StartView(message)).await; - assert_eq!( - consensus.commit_min(), - applied, - "unproven entries cannot be applied" + shard + .on_message(MessageBag::StartView(message.clone())) + .await; + assert_eq!(consensus.commit_min(), applied); + assert!( + journal + .handle() + .header(usize::try_from(applied + 1).unwrap()) + .is_some() ); - assert_eq!( - journal.handle().header(first_unapplied).is_some(), - fail_truncation, + assert!( + journal + .handle() + .header(usize::try_from(old_head).unwrap()) + .is_some() ); - assert_eq!(consensus.view_log_unproven(), fail_truncation); - if !fail_truncation { - return; - } + assert!(consensus.view_log_unproven()); + assert!(!consensus.view_log_entry_proven(applied + 1)); + outbox.drain(); - shard - .on_message(MessageBag::RequestPrepares(peer_request)) - .await; + let request = + Message::<iggy_binary_protocol::RequestPreparesHeader>::new(size_of::< + iggy_binary_protocol::RequestPreparesHeader, + >( + )) + .transmute_header( + |_, header: &mut iggy_binary_protocol::RequestPreparesHeader| { + header.command = Command::RequestPrepares; + header.cluster = consensus.cluster(); + header.group = METADATA_GROUP; + header.replica = 0; + header.view = NEW_VIEW; + header.from_op = applied + 1; + header.to_op = old_head; + header.size = u32::try_from(std::mem::size_of_val(header)).unwrap(); + header.seal(); + }, + ); + shard.on_message(MessageBag::RequestPrepares(request)).await; assert!( outbox.drain().is_empty(), - "unproven entries cannot serve peer repair" + "unproved entries cannot serve peer repair" ); - for _ in 0..consensus::TimeoutManager::NORMAL_HEARTBEAT_TICKS { - consensus.tick(consensus::PlaneKind::Metadata); + if fail_truncation { + shard.on_message(MessageBag::StartView(message)).await; + assert_eq!(consensus.view(), NEW_VIEW); + assert!( + journal + .handle() + .header(usize::try_from(applied + 1).unwrap()) + .is_none(), + "the same-view retry must remove the proven mismatch" + ); + assert!( + journal + .handle() + .header(usize::try_from(old_head).unwrap()) + .is_some(), + "removing a mismatch must preserve the canonical tail" + ); + } + for prepare in prepares.iter().rev() { + let repaired = prepare.clone().transmute_header( + |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { + repaired.0 = header; + repaired.0.command = Command::RepairPrepare; + }, + ); + shard.on_message(MessageBag::RepairPrepare(repaired)).await; + } + assert!(!consensus.view_log_unproven()); + metadata.commit_journal().await; + assert_eq!(consensus.commit_min(), suffix_head.op); + for prepare in prepares { + assert_eq!( + journal + .handle() + .header(usize::try_from(prepare.header().op).unwrap()) + .unwrap() + .checksum, + prepare.header().checksum + ); } - assert_eq!(consensus.status(), Status::ViewChange); - assert_eq!(consensus.view(), NEW_VIEW + 1); - let done = Message::<iggy_binary_protocol::RepairRangeReplyHeader>::new(size_of::< - iggy_binary_protocol::RepairRangeReplyHeader, - >( - )) - .transmute_header( - |_, header: &mut iggy_binary_protocol::RepairRangeReplyHeader| { - header.command = Command::RepairDone; - header.cluster = consensus.cluster(); - header.group = METADATA_GROUP; - header.replica = source.replica(); - header.view = repair_request.view; - header.nonce = repair_request.nonce; - header.op = old_head; - header.size = u32::try_from(std::mem::size_of_val(header)).unwrap(); - header.seal(); - }, - ); - shard.on_message(MessageBag::RepairRangeReply(done)).await; - assert_eq!( - consensus.commit_min(), - applied, - "an old RepairDone cannot apply the suffix left by failed truncation", - ); - assert!(consensus.view_log_unproven()); }); } } diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs index cd60ecb5b..881227b3b 100644 --- a/core/simulator/src/workload/mod.rs +++ b/core/simulator/src/workload/mod.rs @@ -2735,6 +2735,65 @@ mod tests { ); } + #[test] + fn given_a_recreated_fixture_when_offsets_commit_while_settling_should_retain_their_effects() { + let fixture = IggyNamespace::new(1, 1, 0); + let toggler = bound_client(1); + let writer = bound_client(2); + let mut workload = Workload::new(WorkloadOptions::new(0x4C43_4259, 1, vec![fixture])); + let delete = sample_toggle(&mut workload, &toggler); + workload.options.weights = ActionWeights::new(&[(Action::SendMessages, 100)]); + let (_, old_send) = workload.build_request(&writer).unwrap(); + let committed = ApplyReply::default().to_reply_body(); + assert!( + workload + .on_reply(&reply_to(&delete, 0, &committed)) + .is_empty() + ); + let create = sample_toggle(&mut workload, &toggler); + assert!( + workload + .on_reply(&reply_to(&create, 0, &committed)) + .is_empty() + ); + assert!(!workload.shadow.fixtures_restored()); + assert!(workload.on_reply(&reply_to(&old_send, 0, &[])).is_empty()); + assert_eq!(workload.shadow.sends_committed(fixture), 0); + + workload.options.weights = ActionWeights::new(&[(Action::JoinFixtureGroup, 100)]); + workload.shadow.enable_fixture_groups(); + for _ in 0..2 { + let (_, join) = workload.build_request(&writer).unwrap(); + assert!( + workload + .on_reply(&reply_to(&join, 0, &committed)) + .is_empty() + ); + } + workload.options.weights = ActionWeights::new(&[(Action::StoreConsumerOffset, 100)]); + let (_, store) = workload.build_request(&writer).unwrap(); + let (request, _) = StoreConsumerOffsetRequest::decode(store.body()).unwrap(); + assert!(workload.on_reply(&reply_to(&store, 0, &[])).is_empty()); + assert_eq!( + workload.shadow.consumer_offset( + fixture, + request.consumer.kind, + request.consumer.id.as_u32().unwrap() + ), + Some(request.offset), + "a store on the recreated history must survive before every replica settles", + ); + workload.shadow.settle_fixtures(|_, _| true); + assert_eq!( + workload.shadow.consumer_offset( + fixture, + request.consumer.kind, + request.consumer.id.as_u32().unwrap() + ), + Some(request.offset), + ); + } + /// A send sampled while a fixture delete is in flight can commit on the /// deleted incarnation too, even though no replacement starts after it. #[test] diff --git a/core/simulator/src/workload/shadow.rs b/core/simulator/src/workload/shadow.rs index eb961daa9..772a5350d 100644 --- a/core/simulator/src/workload/shadow.rs +++ b/core/simulator/src/workload/shadow.rs @@ -96,6 +96,8 @@ pub enum ToggleFate { pub struct Replacements { /// Deletes sampled, less those refused before they reached the server. pub started: u64, + /// Recreated topics whose successful create has been observed. + pub created: u64, /// Replacements over: every live replica hosts the recreated incarnation. pub settled: u64, } @@ -456,7 +458,10 @@ impl Shadow { } } } - FixtureState::Settling { .. } => self.namespaces_live.extend(namespaces), + FixtureState::Settling { .. } => { + self.namespaces_live.extend(namespaces); + self.fixtures[index].replacements.created += 1; + } _ => {} } } @@ -509,19 +514,20 @@ impl Shadow { /// A pending delete can still refuse, so retain committed effects until its /// success clears them. After a successful delete, late replies belong to the - /// old history until the replacement settles. + /// old history once the replacement has been created, including while it settles. #[must_use] pub fn effect_is_current(&self, namespace: IggyNamespace, sampled: Replacements) -> bool { self.fixtures .iter() .find(|fixture| fixture.namespaces.contains(&namespace)) .is_none_or(|fixture| { - sampled.settled == fixture.replacements.settled + sampled.created == fixture.replacements.created && matches!( fixture.state, FixtureState::Live | FixtureState::Deleting { .. } | FixtureState::DeleteUnknown + | FixtureState::Settling { .. } ) }) } @@ -533,6 +539,7 @@ impl Shadow { .iter() .fold(Replacements::default(), |total, fixture| Replacements { started: total.started + fixture.replacements.started, + created: total.created + fixture.replacements.created, settled: total.settled + fixture.replacements.settled, }) } diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go index d054423d0..efb45f556 100644 --- a/foreign/go/client/tcp/tcp_core.go +++ b/foreign/go/client/tcp/tcp_core.go @@ -856,8 +856,7 @@ func (c *IggyTcpClient) attempt( } if captured, ok := ctx.Value(capturedPartitionContext{}).(iggcon.PartitionContext); ok { vsr.StampPartitionContext((*[vsr.HeaderSize]byte)(frame[:vsr.HeaderSize]), captured) - } - if _, captured := ctx.Value(capturedPartitionContext{}).(iggcon.PartitionContext); !captured { + } else { vsr.StampPartitionContext((*[vsr.HeaderSize]byte)(frame[:vsr.HeaderSize]), iggcon.PartitionContext{MetadataOp: c.metadataWatermark.Load()}) } stamped = true
