github-actions[bot] commented on code in PR #4433:
URL: https://github.com/apache/iggy/pull/4433#discussion_r4272906492
##########
core/shard/src/lib.rs:
##########
@@ -4947,11 +4947,19 @@ where
return;
};
partition.ensure_materialization_recovery();
+ let prior_log_view = partition.consensus().log_view();
let actions =
partition
.consensus()
.handle_start_view(PlaneKind::Partitions, &header,
suffix_body);
- if !partition.consensus().is_transferring() &&
partition.consensus().view_log_unproven() {
+ let adopted = !actions.is_empty();
+ // Consumed on every adoption, so a restart's flag never outlives the
first.
+ let recovered = adopted && partition.take_recovered_unproven();
Review Comment:
warning: A transferring adoption consumes the boot flag while `fence_below`
stays false, so no fence installs and a later abort leaves the recovered suffix
unfenced. Consume the flag only where the fence installs.
##########
core/partitions/src/view_fence.rs:
##########
@@ -0,0 +1,155 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Entries a replica journaled before adopting a newer view, held back from
+//! commit and acknowledgement until the view's log proves them.
+
+use iggy_binary_protocol::{CHECKSUM_UNSEALED, PrepareHeader};
+
+/// An unproven window `(floor, through]` installed in `view` and, once known,
that
+/// view's header at `through + 1`, whose `parent` the window must chain into.
The
+/// floor is `IggyPartition::fence_floor`: the higher of `commit_min` and the
+/// commit point recovered from the WAL, which the fenced walk may still reach.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ViewFence {
+ pub view: u32,
+ pub through: u64,
+ pub anchor: Option<PrepareHeader>,
+ /// The op the last proof stopped at for lack of an entry. Until it arrives
+ /// nothing below it can change, so a proof does not rescan the window.
+ pub hole: Option<u64>,
+ /// Ticks spent without an anchor, paced by
[`crate::IggyPartition::view_fence_probe_due`].
+ pub unanchored_ticks: u32,
+}
+
+/// What the `parent` chain says about a fenced window.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ViewWindowVerdict {
+ /// Every op in the window links into the anchor.
+ Proven,
+ /// No entry at `op`, so nothing at or below it can be linked yet.
+ Hole { op: u64 },
+ /// The entry at `op` belongs to a log the view discarded: it is not the
one
+ /// the op above chains to, or it is not the copy the WAL flush would
write.
+ Break { op: u64 },
+}
+
+/// Walk `(floor, through]` top-down from `anchor_parent`, the checksum the
view's
+/// entry at `through` must carry.
+///
+/// An unsealed checksum on either side counts as a link, as in the shard's
+/// `header_is_view_entry`: a pre-seal WAL entry carries no identity to
compare.
+#[must_use]
+pub fn view_window_verdict(
+ floor: u64,
+ through: u64,
+ anchor_parent: u128,
+ header_at: impl Fn(u64) -> Option<PrepareHeader>,
+) -> ViewWindowVerdict {
+ let mut expected = anchor_parent;
+ for op in (floor + 1..=through).rev() {
+ let Some(local) = header_at(op) else {
+ return ViewWindowVerdict::Hole { op };
+ };
+ let linked = local.checksum == expected
Review Comment:
warning: `view_window_verdict` treats an unsealed local entry as a link,
then adopts its zero `parent` for every op below it, so the window proves on
entries with no identity. The anchor filters at
`core/partitions/src/iggy_partition.rs:8864` refuse that header for that
reason, so return `Break` for an unsealed entry.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8749,6 +8841,346 @@ where
})
}
+ /// Hold `(fence_floor, through]` back from commit and acknowledgement,
where
+ /// `fence_floor` is the higher of `commit_min` and the recovered commit
point, until
+ /// [`Self::prove_view_window`] links it into `anchor`, the current view's
header
+ /// at `through + 1`. Without an anchor the first current-view prepare at
that op
+ /// supplies one (`offer_view_anchor`).
+ ///
+ /// For a replica that adopted a newer view: it can hold an entry from a
log the
+ /// view discarded under the view's commit point, which nothing else
compares
+ /// against the view's log.
+ pub fn fence_view_window(&mut self, through: u64, anchor:
Option<PrepareHeader>) {
+ if through <= self.fence_floor() {
+ self.view_fence = None;
+ return;
+ }
+ let view = self.consensus.view();
+ let previous = self
+ .view_fence
+ .filter(|fence| fence.view == view && fence.through == through);
+ // An unsealed header names no identity, and a zero `parent` above op 1
+ // would link anything below it.
+ let anchor = anchor
+ .filter(|anchor| {
+ anchor.checksum != CHECKSUM_UNSEALED
+ && (anchor.parent != CHECKSUM_UNSEALED || anchor.op == 1)
+ })
+ .or_else(|| previous.and_then(|fence| fence.anchor));
+ if previous.is_none() {
+ info!(
+ namespace_raw = self.namespace().inner(),
+ view,
+ commit_min = self.consensus.commit_min(),
+ floor = self.fence_floor(),
+ recovered_commit = self.recovered_commit,
+ through,
+ anchored = anchor.is_some(),
+ "holding partition entries journaled before this view until
its log proves them"
+ );
+ }
+ self.view_fence = Some(ViewFence {
+ view,
+ through,
+ anchor,
+ hole: previous.and_then(|fence| fence.hole),
+ unanchored_ticks: previous.map_or(0, |fence|
fence.unanchored_ticks),
+ });
+ }
+
+ /// Highest op through which this replica holds every entry above its
commit
+ /// point. Repair progress, since a fenced commit point cannot move until
the
+ /// whole window is held.
+ #[must_use]
+ pub fn held_through(&self) -> u64 {
+ self.log
+ .journal()
+ .inner
+ .held_through(self.consensus.commit_min())
+ }
+
+ /// Whether an anchored fence holds every entry of its window, so a proof
+ /// now concludes instead of waiting on repair.
+ #[must_use]
+ pub fn view_fence_held(&self) -> bool {
+ self.view_fence.as_ref().is_some_and(|fence| {
+ fence.anchor.is_some()
+ && fence.through > self.consensus.commit_min()
+ && self.log.journal().inner.holds_op(fence.through)
Review Comment:
simplification: `holds_op(fence.through)` repeats what `held_through() >=
fence.through` already proves: a contiguous resident run from `commit_min`
through that op. Drop the index lookup.
##########
core/consensus/src/impls.rs:
##########
@@ -3537,8 +3537,19 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
/// Falls back to the announced `op` on an empty body (probe answer,
stale-view
/// correction).
fn adopt_start_view_suffix(&self, header: &StartViewHeader, suffix_body:
&[u8]) -> u64 {
- let suffix = match dvc_suffix_decode(suffix_body, header.op, 0, 0) {
- Ok(suffix) => suffix,
+ match MergedLog::from_start_view(header, suffix_body) {
+ Ok(Some(log)) => *self.pending_view_log.borrow_mut() = Some(log),
+ // A numbers-only StartView still has to replace the pending log
while
+ // the held history is unproven, so the proof waits on this view's
head.
+ Ok(None) if self.view_log_unproven() => {
+ *self.pending_view_log.borrow_mut() = Some(MergedLog {
Review Comment:
simplification: `MergedLog::from_start_view` declines the numbers-only body
with `Ok(None)`, and `adopt_start_view_suffix` rebuilds the same value field by
field. Return the numbers-only log and let the callers test
`headers.is_empty()`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -5848,7 +5902,12 @@ where
else {
unreachable!("truncate transition exists");
};
- self.remove_resident_range(ops).await?;
+ let unproven_drop = self.unproven_drop_from.take() ==
Some(*ops.start());
Review Comment:
nit: The truncate step takes `unproven_drop_from` before it compares the
range start, so a cut for another range clears the marker and skips
`note_unproven_dropped`. Compare `*ops.start()` first and take the marker only
on a match.
##########
core/shard/src/lib.rs:
##########
@@ -11719,6 +11830,98 @@ async fn reconcile_partition_view_divergence<B, SB>(
}
}
+/// The lowest header in a view's suffix that can anchor a fence. A blank
slot, or
+/// an unsealed one, names no identity: its zero `parent` would link anything
+/// below it. Op 1 is the one header whose real parent is zero.
+fn view_fence_anchor(headers: &[PrepareHeader]) -> Option<PrepareHeader> {
+ headers
+ .iter()
+ .filter(|header| {
+ matches!(dvc_header_kind(header), DvcHeaderKind::Valid)
+ && header.checksum != CHECKSUM_UNSEALED
Review Comment:
simplification: The same sealed-header test is written out three times
(`core/partitions/src/iggy_partition.rs:8866`, `:8989`, here), so a change to
one copy can drift from the others. Move `checksum != CHECKSUM_UNSEALED &&
(parent != CHECKSUM_UNSEALED || op == 1)` onto `PrepareHeader` and call it from
all three.
##########
core/simulator/src/stale_prepare_commit_tests.rs:
##########
@@ -0,0 +1,538 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! A replica must never commit a prepare the cluster did not commit.
+//!
+//! The old primary journals an op, is cut off before any peer sees it, and the
+//! other two replicas elect a new view that commits a DIFFERENT prepare at the
+//! same op. Whether the old primary rejoins that view or is later elected over
+//! it, its stale prepare sits below the view's canonical headers, and it must
+//! end up serving the view's entry at that offset, not its own.
+
+use bytes::Bytes;
+use consensus::{PartitionsHandle, Status};
+use futures::FutureExt;
+use iggy_common::PollingStrategy;
+use partitions::{PollingArgs, PollingConsumer};
+use server_common::send_messages::{BatchIntegrity, decode_batch_slice_with};
+use server_common::sharding::IggyNamespace;
+
+use crate::Simulator;
+use crate::client::SimClient;
+use crate::packet::{PacketSimulatorOptions, ProcessId};
+
+const REPLICAS: u8 = 3;
+const CLIENT_ID: u128 = 1;
+const OLD_PRIMARY: u8 = 0;
+const STEPS_PER_SEND: usize = 12;
+/// Well past `NORMAL_HEARTBEAT_TICKS` (500), so the majority elects.
+const ELECTION_STEPS: usize = 5_000;
+const SETTLE_STEPS: usize = 5_000;
+/// Long enough for a lone survivor's view changes to rotate onto the old
primary.
+const REJOIN_STEPS: usize = 40_000;
+
+fn cluster(seed: u64) -> (Simulator, SimClient) {
Review Comment:
simplification: `cluster` and `group_state` copy the helpers in
`partition_repair_driver_tests` (`core/simulator/src/lib.rs:8926`). Import them
and make `group_state` `pub`, as that module's other helpers are.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8371,6 +8442,13 @@ where
.await;
return;
}
+ // After the in-flight commit: a proof that breaks truncates the
journal.
+ if self.view_fenced() {
+ self.prove_view_window().await;
Review Comment:
warning: `prove_view_window` re-runs on every replicated prepare while the
fence is broken: `drop_unproven_from` returns unchanged when history is busy,
and `core/shard/src/lib.rs:3382` calls `commit_journal` per prepare. Cache the
broken op on the fence and skip the walk while it stays unchanged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
