krishvishal commented on code in PR #4073:
URL: https://github.com/apache/iggy/pull/4073#discussion_r3957104981
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -482,15 +510,33 @@ where
/// revalidates that the head is still this exact entry before popping and
/// applying it. A driver dropped at an await strands nothing; a sibling driver
/// that committed the op first fails the caller's revalidation and re-peeks.
+///
+/// Bounded below for the reason [`drain_committable_prefix`] is, and reported
+/// rather than asserted for the same one. Holding is safe: a shard pump's
panic is
+/// swallowed by `compio::runtime::spawn`, while `tick_metadata` re-arms
repair on
+/// the level.
pub fn peek_committable_head<B, P>(consensus: &VsrConsensus<B, P>) ->
Option<PrepareHeader>
where
B: MessageBus,
P: Pipeline<Entry = PipelineEntry>,
{
let commit = consensus.commit_max();
- consensus
+ let next = consensus.commit_min() + 1;
+ let head = consensus
.pipeline_head_header()
- .filter(|header| header.op <= commit)
+ .filter(|header| header.op <= commit)?;
+ if head.op != next {
Review Comment:
Fixed. Both walks capped at the pipeline head, with an absent head meaning
no cap. `given_an_empty_pipeline_when_walking_the_journal_should_not_cap` pins
that a backup still walks.
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -446,18 +456,35 @@ where
P: Pipeline<Entry = PipelineEntry>,
{
let commit = consensus.commit_max();
+ let commit_min = consensus.commit_min();
+ let replica = consensus.replica();
let mut drained = Vec::new();
consensus.with_pipeline_mut(|pipeline| {
+ let mut next = commit_min + 1;
while let Some(head_op) = pipeline.head().map(|entry| entry.header.op)
{
if head_op > commit {
break;
}
+ if head_op != next {
Review Comment:
Fixed. `collect_committable_from_journal` lowers its ceiling to `head.op -
1` for a head at or above `from_op`, so the journal walk can no longer
overshoot the drain and strand the reply.
##########
core/shard/src/lib.rs:
##########
@@ -5550,9 +5592,12 @@ where
// from the evicted ring or the flushed segments.
let missing = {
let journal = partition.log.journal();
- first_op_not_covered(&pending, consensus.commit_min(), |op| {
- journal.inner.header_by_op(op)
- })
+ first_op_not_covered(
+ &pending,
+ consensus.commit_min(),
+ consensus.commit_min(),
Review Comment:
Fixed. Ring+resident probe in one pass, which also kills the quadratic
`header_by_op`, plus a real fill path: `request_partition_view_repair` arms
from the DVC senders that named the op.
##########
core/shard/src/lib.rs:
##########
@@ -5023,6 +5026,9 @@ where
if !in_scope {
return;
}
+ // The serving peer is answering. Clears the stall clock and the
budget
+ // so a window served in chunks cannot rotate off a healthy stream.
+ self.note_metadata_repair_progress();
Review Comment:
Fixed. Reset restored below all three returns, and the budget is now cleared
only by `note_metadata_repair_walked`, never by a bare frame, so it accumulates
across rounds.
--
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]
