github-actions[bot] commented on code in PR #4483:
URL: https://github.com/apache/iggy/pull/4483#discussion_r4272780911
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -11929,6 +11908,155 @@ where
}
}
+ /// Refuse a floor whose window does not connect to recovered durable
state,
+ /// or `None` when the walk below makes the floor moot and completion
+ /// proceeds as without one.
+ async fn refuse_disconnected_floor(
+ &mut self,
+ config: &PartitionsConfig,
+ session: RepairSession,
+ floor: u64,
+ durable_end: Option<u64>,
+ anchor: Option<RepairAnchor>,
+ window_complete: bool,
+ ) -> Option<RepairConclusion> {
+ self.commit_journal(config).await;
+ // Ops this replica already held can carry the walk past the floor.
+ // Under lazy flush their headers stay resident, so the leading edge
+ // below would read as resident and refuse a floor that no longer
+ // moves `commit_min`, spending a state transfer on nothing. Judged by
+ // the peer's unclamped floor: one clamped down to `commit_max` still
+ // names ops the peer evicted.
+ if session
+ .floor
+ .is_some_and(|raw_floor| self.consensus().commit_min() >=
raw_floor)
+ {
+ return None;
+ }
+ let leading_edge_resident = self.repair_leading_edge_resident(floor,
anchor);
+ // Definitive once the window, or its leading edge `(floor,
anchor.op)`,
+ // is resident: until then a frame still to come can land below the
+ // anchor and connect. The leading edge suffices because the primary
+ // assigns offsets in op order.
+ if window_complete || leading_edge_resident {
+ let reason = if window_complete {
+ "whole_window"
+ } else {
+ "leading_edge"
+ };
+ tracing::error!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ floor,
+ commit_to_op = session.commit_to_op,
+ anchor_op = ?anchor.map(|anchor| anchor.op),
+ anchor_base_offset = ?anchor.map(|anchor| anchor.base_offset),
+ reason,
+ recovered_durable_offset = ?durable_end,
+ "refusing commit floor: repaired window does not connect \
+ to recovered durable state (needs state transfer)"
+ );
+ self.repair = None;
+ return Some(RepairConclusion::FloorRefused {
+ floor,
+ to_op: session.commit_to_op,
+ });
+ }
+ tracing::warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ floor,
+ commit_to_op = session.commit_to_op,
+ anchor_op = ?anchor.map(|anchor| anchor.op),
+ anchor_base_offset = ?anchor.map(|anchor| anchor.base_offset),
+ recovered_durable_offset = ?durable_end,
+ "commit floor does not connect yet; repairing below the anchor"
+ );
+ Some(RepairConclusion::InProgress)
+ }
+
+ /// Where the next pull of the repair window starts: past the resident
+ /// prefix above a floor that still holds the walk, else `commit_min + 1`.
+ /// `None` without a session.
+ ///
+ /// A held walk never moves `commit_min`, so pulling from it re-serves the
+ /// same first chunk forever while the rest of the window stays absent.
+ #[must_use]
+ pub fn repair_pull_from(&self) -> Option<u64> {
+ let session = self.repair.as_ref()?;
+ let commit_min = self.consensus().commit_min();
+ let Some(floor) = session.floor.filter(|&floor| floor > commit_min)
else {
+ return Some(commit_min.saturating_add(1));
+ };
+ let journal = &self.log.journal().inner;
+ let mut op = floor.saturating_add(1);
+ while op <= session.fetch_to_op && journal.holds_op(op) {
+ op += 1;
+ }
+ Some(op)
+ }
+
+ /// The next chunk to pull right after a `RepairDone` while a floor holds
+ /// the walk, or `None` when the resident prefix has not grown since the
+ /// last such pull or reaches past `fetch_to_op`.
+ pub fn next_floor_pull(&mut self) -> Option<u64> {
+ let commit_min = self.consensus().commit_min();
+ let from_op = self.repair_pull_from()?;
+ let session = self.repair.as_mut()?;
+ let floor_pending = session.floor.is_some_and(|floor| floor >
commit_min);
+ if !floor_pending || from_op > session.fetch_to_op || from_op <=
session.floor_pulled_from {
+ return None;
+ }
+ session.floor_pulled_from = from_op;
+ Some(from_op)
+ }
+
+ /// The lowest resident `SendMessages` batch in `(floor, commit_to_op]`:
+ /// where the repair window begins in offset space.
+ ///
+ /// Read from the journal rather than tracked per session. A batch at or
+ /// below the floor is not what the floor claims recovered state stands in
+ /// for, and a batch resident from an earlier session or the boot WAL
+ /// counts as much as one repaired now.
+ fn repair_window_anchor(&self, floor: u64, commit_to_op: u64) ->
Option<RepairAnchor> {
+ let from = floor.checked_add(1)?;
+ let (op, entry) = self
+ .log
+ .journal()
+ .inner
+ .lowest_resident_send_in(from..=commit_to_op)?;
+ let batch = decode_prepare_slice(entry.as_slice())
Review Comment:
warning: `repair_window_anchor` decodes the batch with the
checksum-validating `decode_prepare_slice`, then reads only `base_offset` from
it. Use `decode_prepare_slice_trusted`, as `committed_batch_stats_for_prepare`
does, because the batch was validated before it was journaled.
##########
core/shard/src/lib.rs:
##########
@@ -8091,10 +8134,8 @@ where
);
continue;
}
- let repair_from = partition
- .consensus()
- .view_log_repair_op()
- .unwrap_or_else(|| commit_min.saturating_add(1));
+ let proof_from = partition.consensus().view_log_repair_op();
+ let pull_from = partition.repair_pull_from();
Review Comment:
warning: `tick_partitions` calls `repair_pull_from()` on every tick for each
live repair session, yet only the stall retry uses the value. Compute it only
on the tick that retry is due, because the resident-prefix walk repeats on each
tick.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -11929,6 +11908,155 @@ where
}
}
+ /// Refuse a floor whose window does not connect to recovered durable
state,
+ /// or `None` when the walk below makes the floor moot and completion
+ /// proceeds as without one.
+ async fn refuse_disconnected_floor(
+ &mut self,
+ config: &PartitionsConfig,
+ session: RepairSession,
+ floor: u64,
+ durable_end: Option<u64>,
+ anchor: Option<RepairAnchor>,
+ window_complete: bool,
+ ) -> Option<RepairConclusion> {
+ self.commit_journal(config).await;
+ // Ops this replica already held can carry the walk past the floor.
+ // Under lazy flush their headers stay resident, so the leading edge
+ // below would read as resident and refuse a floor that no longer
+ // moves `commit_min`, spending a state transfer on nothing. Judged by
+ // the peer's unclamped floor: one clamped down to `commit_max` still
+ // names ops the peer evicted.
+ if session
Review Comment:
warning: `refuse_disconnected_floor` walks once and then tests `commit_min
>= session.floor`, but `commit_journal` stops its journal half at
`COMMIT_WALK_OPS_MAX` (64). A longer resident run keeps the floor looking
unmooted, so a complete window refuses and spends a state transfer. Walk again
while the pass advances before judging the floor.
##########
core/partitions/src/types.rs:
##########
@@ -311,6 +311,10 @@ pub struct RepairSession {
/// Ticks since the stream last made progress; at
/// [`REPAIR_RETRY_TICKS`] the remaining window is re-requested.
pub idle_ticks: u32,
+ /// Start of the last chunk pulled above a pending [`Self::floor`]. A
+ /// `RepairDone` pulls again only past it, so a peer that serves nothing
+ /// is left to the stall retry instead of answering an unthrottled loop.
+ pub floor_pulled_from: u64,
Review Comment:
nit: The doc does not state the 0 sentinel, and the stall retry sets only
`from_op`. Say in the doc that only a `RepairDone` pull writes
`floor_pulled_from` and that 0 means no pull yet.
##########
core/shard/src/lib.rs:
##########
@@ -10522,7 +10580,7 @@ where
match gap_repair_peer(
consensus.replica(),
consensus.replica_count(),
- consensus.primary_index(consensus.view()),
+ Some(consensus.primary_index(consensus.view())),
Review Comment:
warning: `gap_repair_peer`'s doc says the two planes owe the same answer,
but the three metadata sites still pass `primary_index(consensus.view())`, and
no call records a metadata primary. Record the metadata primary and pass the
hint at lines 6475, 10583 and 10688.
##########
core/shard/src/lib.rs:
##########
@@ -2840,23 +2840,49 @@ where
)
}
+/// Whether a transfer descriptor comes from a replica that knows less than
this
+/// one. The serving view is compared against `log_view`, not `view`: a replica
+/// that needs state transfer cannot join a view change, so its `view` runs
ahead.
+const fn transfer_offer_is_behind<B, P>(
Review Comment:
simplification: `transfer_offer_is_behind` takes `&VsrConsensus<B, P>` and
two generic bounds to compare two fields. Simpler: pass `local_log_view` and
`local_commit_max` as scalars, which the caller already holds at line 9581, and
the test fixture loses its consensus. Saves: about 20 lines.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -11929,6 +11908,155 @@ where
}
}
+ /// Refuse a floor whose window does not connect to recovered durable
state,
+ /// or `None` when the walk below makes the floor moot and completion
+ /// proceeds as without one.
+ async fn refuse_disconnected_floor(
+ &mut self,
+ config: &PartitionsConfig,
+ session: RepairSession,
+ floor: u64,
+ durable_end: Option<u64>,
+ anchor: Option<RepairAnchor>,
+ window_complete: bool,
+ ) -> Option<RepairConclusion> {
+ self.commit_journal(config).await;
+ // Ops this replica already held can carry the walk past the floor.
+ // Under lazy flush their headers stay resident, so the leading edge
+ // below would read as resident and refuse a floor that no longer
+ // moves `commit_min`, spending a state transfer on nothing. Judged by
+ // the peer's unclamped floor: one clamped down to `commit_max` still
+ // names ops the peer evicted.
+ if session
+ .floor
+ .is_some_and(|raw_floor| self.consensus().commit_min() >=
raw_floor)
+ {
+ return None;
+ }
+ let leading_edge_resident = self.repair_leading_edge_resident(floor,
anchor);
+ // Definitive once the window, or its leading edge `(floor,
anchor.op)`,
+ // is resident: until then a frame still to come can land below the
+ // anchor and connect. The leading edge suffices because the primary
+ // assigns offsets in op order.
+ if window_complete || leading_edge_resident {
+ let reason = if window_complete {
+ "whole_window"
+ } else {
+ "leading_edge"
+ };
+ tracing::error!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ floor,
+ commit_to_op = session.commit_to_op,
+ anchor_op = ?anchor.map(|anchor| anchor.op),
+ anchor_base_offset = ?anchor.map(|anchor| anchor.base_offset),
+ reason,
+ recovered_durable_offset = ?durable_end,
+ "refusing commit floor: repaired window does not connect \
+ to recovered durable state (needs state transfer)"
+ );
+ self.repair = None;
+ return Some(RepairConclusion::FloorRefused {
+ floor,
+ to_op: session.commit_to_op,
+ });
+ }
+ tracing::warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ floor,
+ commit_to_op = session.commit_to_op,
+ anchor_op = ?anchor.map(|anchor| anchor.op),
+ anchor_base_offset = ?anchor.map(|anchor| anchor.base_offset),
+ recovered_durable_offset = ?durable_end,
+ "commit floor does not connect yet; repairing below the anchor"
+ );
+ Some(RepairConclusion::InProgress)
+ }
+
+ /// Where the next pull of the repair window starts: past the resident
+ /// prefix above a floor that still holds the walk, else `commit_min + 1`.
+ /// `None` without a session.
+ ///
+ /// A held walk never moves `commit_min`, so pulling from it re-serves the
+ /// same first chunk forever while the rest of the window stays absent.
+ #[must_use]
+ pub fn repair_pull_from(&self) -> Option<u64> {
+ let session = self.repair.as_ref()?;
+ let commit_min = self.consensus().commit_min();
+ let Some(floor) = session.floor.filter(|&floor| floor > commit_min)
else {
+ return Some(commit_min.saturating_add(1));
+ };
+ let journal = &self.log.journal().inner;
+ let mut op = floor.saturating_add(1);
+ while op <= session.fetch_to_op && journal.holds_op(op) {
+ op += 1;
+ }
+ Some(op)
+ }
+
+ /// The next chunk to pull right after a `RepairDone` while a floor holds
+ /// the walk, or `None` when the resident prefix has not grown since the
+ /// last such pull or reaches past `fetch_to_op`.
+ pub fn next_floor_pull(&mut self) -> Option<u64> {
Review Comment:
nit: `next_floor_pull` writes `session.floor_pulled_from` although its name
reads as a query. Name it for the write, or state the write in the doc.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -11929,6 +11908,155 @@ where
}
}
+ /// Refuse a floor whose window does not connect to recovered durable
state,
+ /// or `None` when the walk below makes the floor moot and completion
+ /// proceeds as without one.
+ async fn refuse_disconnected_floor(
+ &mut self,
+ config: &PartitionsConfig,
+ session: RepairSession,
+ floor: u64,
+ durable_end: Option<u64>,
+ anchor: Option<RepairAnchor>,
+ window_complete: bool,
+ ) -> Option<RepairConclusion> {
+ self.commit_journal(config).await;
+ // Ops this replica already held can carry the walk past the floor.
+ // Under lazy flush their headers stay resident, so the leading edge
+ // below would read as resident and refuse a floor that no longer
+ // moves `commit_min`, spending a state transfer on nothing. Judged by
+ // the peer's unclamped floor: one clamped down to `commit_max` still
+ // names ops the peer evicted.
+ if session
+ .floor
+ .is_some_and(|raw_floor| self.consensus().commit_min() >=
raw_floor)
+ {
+ return None;
+ }
+ let leading_edge_resident = self.repair_leading_edge_resident(floor,
anchor);
+ // Definitive once the window, or its leading edge `(floor,
anchor.op)`,
+ // is resident: until then a frame still to come can land below the
+ // anchor and connect. The leading edge suffices because the primary
+ // assigns offsets in op order.
+ if window_complete || leading_edge_resident {
+ let reason = if window_complete {
+ "whole_window"
+ } else {
+ "leading_edge"
+ };
+ tracing::error!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ floor,
Review Comment:
nit: The refusal diagnostic logs the clamped floor, so the operator reads a
value below the peer's report. Log `session.floor` as a separate field.
##########
core/shard/src/lib.rs:
##########
@@ -5099,6 +5125,14 @@ where
}
let config = planes.1.0.config();
+ // Before the view filter: a replica whose view climbed through
+ // elections nobody joined drops the real primary's lower-view
+ // heartbeats, and they are how it learns whom to repair from.
+ if let Some(partition) =
planes.1.0.get_by_ns(&IggyNamespace::from_raw(header.group)) {
Review Comment:
nit: The Commit handler records the primary hint before the incarnation
check, while the descriptor ingress checks it first. Move the record after the
incarnation check and keep it above the view filter.
##########
core/shard/src/lib.rs:
##########
@@ -8173,12 +8222,12 @@ where
continue;
};
let consensus = partition.consensus();
- let primary = consensus.primary_index(consensus.view());
+ consensus.forget_primary(peer);
let next_peer = next_transfer_peer(
consensus.replica(),
peer,
consensus.replica_count(),
- primary,
+ consensus.primary_hint(),
Review Comment:
nit: `spent_repair_ring_transfer_peer` picks the ring-spent conversion's
target with `primary_index(consensus.view())`, which arms the transfer at line
8213 against a replica the group may not follow. Prefer `primary_hint()` there,
and keep the view math as the fallback for an unheard group.
--
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]
