numinnex commented on code in PR #3824:
URL: https://github.com/apache/iggy/pull/3824#discussion_r3726705675
##########
core/shard/src/lib.rs:
##########
@@ -5057,10 +5523,387 @@ where
incarnation: 0,
target: None,
namespace: consensus.namespace(),
+ // Correcting a peer on a stale view, not concluding a view change:
this
+ // publishes the settled frontier, which the peer reaches by repair.
+ suffix: Vec::new(),
};
dispatch_vsr_actions::<B, P, J>(consensus, None, &[action]).await;
}
+/// Rebuild the new primary's pipeline over `from_op..=to_op` from local
journal
+/// headers.
+///
+/// A gap means the caller started the view before its journal could serve the
+/// merged log: a bug in the transition, not a data condition. Nothing is
+/// truncated, because truncating to the last findable op discards ops
committed
+/// on a quorum and already acknowledged. The pipeline is left short, the
commit
+/// walk stalls at the gap, and repair fills it in.
+fn rebuild_pipeline_entries<B, P>(
+ consensus: &VsrConsensus<B, P>,
+ self_id: u8,
+ from_op: u64,
+ to_op: u64,
+ header_at: impl Fn(u64) -> Option<PrepareHeader>,
+) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+{
+ let mut gap_at = None;
+ let entries: Vec<_> = (from_op..=to_op)
+ .map_while(|op| {
+ let header = header_at(op).or_else(|| {
+ gap_at = Some(op);
+ None
+ })?;
+ // Lift the monotonic timestamp floor to the rebuilt log so
+ // post-view-change prepares cannot stamp below committed ones.
+ consensus.observe_prepare_timestamp(header.timestamp);
+ let mut entry = consensus::PipelineEntry::new(header);
+ entry.add_ack(self_id);
+ Some(entry)
+ })
+ .collect();
+
+ if let Some(missing_op) = gap_at {
+ tracing::error!(
+ replica = self_id,
+ missing_op,
+ range_start = from_op,
+ range_end = to_op,
+ rebuilt = entries.len(),
+ "RebuildPipeline: journal gap at op {missing_op} while starting a
view; leaving the \
+ sequencer at {to_op} and stalling the commit walk. Truncating
here would discard ops \
+ the view change proved recoverable."
+ );
+ }
+
+ let mut pipeline = consensus.pipeline().borrow_mut();
+ for entry in entries {
+ pipeline.push(entry);
+ }
+}
+
+/// Snapshot this replica's uncommitted suffix into consensus, if the journal
has
+/// moved since the last snapshot.
+///
+/// Called before every handler that could start or join a view change:
consensus
+/// records its own `DoViewChange` there and has no journal to read. A stale
+/// snapshot is never reused; consensus tags it with its `(op, commit)` and
falls
+/// back to an empty suffix, stalling the view change rather than nacking an op
+/// since acquired.
+fn refresh_metadata_dvc_suffix<B, P, MJ>(consensus: &VsrConsensus<B, P>,
journal: Option<&MJ>)
+where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+{
+ if !consensus.local_dvc_suffix_stale() {
+ return;
+ }
+ let op = consensus.sequencer().current_sequence();
+ let commit = consensus.commit_max().min(op);
+ let pending = adopted_view_headers(consensus);
+ consensus.set_local_dvc_suffix(build_metadata_dvc_suffix(
+ journal,
+ commit,
+ op,
+ pending.as_ref().map(|pending| pending.headers.as_slice()),
+ ));
+}
+
+/// The adopted view's headers, when they describe a log this replica has NOT
itself
+/// decided.
+///
+/// `None` for the primary-elect holding the log its own merge produced: that
log is
+/// a proposal it is still repairing toward and may contain ops a later view
+/// truncated, so stitching it into its own `DoViewChange` would re-assert
them.
+///
+/// A backup's parked log is the opposite: headers the view already decided and
+/// announced, which this replica acknowledged and is repairing to hold.
+fn adopted_view_headers<B, P>(consensus: &VsrConsensus<B, P>) ->
Option<consensus::MergedLog>
+where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+{
+ if consensus.is_primary_for_view(consensus.view()) {
+ return None;
+ }
+ consensus.pending_view_log()
+}
+
+/// Snapshot a partition's uncommitted suffix into its consensus.
+///
+/// Same contract as [`Self::refresh_metadata_dvc_suffix`]. The partition
journal
+/// is in-memory only, so after a restart it reads empty and this replica votes
+/// all-nack: correct, since the ops really are lost and the merge needs a peer
+/// that still holds them.
+fn refresh_partition_dvc_suffix<B>(partition: &partitions::IggyPartition<B>)
+where
+ B: MessageBus,
+{
+ let consensus = partition.consensus();
+ if !consensus.local_dvc_suffix_stale() {
+ return;
+ }
+ let op = consensus.sequencer().current_sequence();
+ let commit = consensus.commit_max().min(op);
+ let journal = partition.log.journal();
+ let pending = adopted_view_headers(consensus);
+ let suffix = build_dvc_suffix(
+ commit,
+ op,
+ |entry_op| journal.inner.header_by_op(entry_op),
+ pending.as_ref().map(|pending| pending.headers.as_slice()),
+ );
+ consensus.set_local_dvc_suffix(suffix);
+}
+
+/// The suffix headers a `DoViewChange` or `StartView` carries, as raw bytes.
+///
+/// `size` is attacker-controlled, so it is clamped to what arrived; a short
read
+/// decodes as a malformed suffix and the DVC is dropped.
+fn control_suffix_body<H>(msg: &Message<H>) -> &[u8]
+where
+ H: iggy_binary_protocol::ConsensusHeader,
+{
+ let slice = msg.as_slice();
+ let start = size_of::<H>();
+ let end = (msg.header().size() as usize).min(slice.len());
+ if end <= start {
+ return &[];
+ }
+ &slice[start..end]
+}
+
+/// Seal a control-message body. Zero for an empty body, which is the unsealed
+/// sentinel every other integrity field in this protocol uses.
+fn control_body_checksum(body: &[u8]) -> u128 {
+ if body.is_empty() {
+ return 0;
+ }
+ u128::from(iggy_common::calculate_checksum(body))
+}
+
+/// The body of a control frame, once it matches the checksum its header
carries.
+///
+/// `None` means corruption in transit and the frame must be dropped whole: the
+/// header numbers describe a body that did not arrive intact, so neither half
is
+/// trustworthy. This is what covers a body-carrying control message end to
end.
+///
+/// Keyed on whether a body is present, NOT on whether `checksum_body` looks
+/// sealed: skipping the check when that field reads zero makes the layer
+/// bypassable by clearing the one field that decides whether anything is
checked.
+/// A peer predating the suffix sends no body, so a non-empty body always came
+/// from a sender that seals it, and a zero checksum there is corruption.
+fn control_suffix_body_verified<H>(msg: &Message<H>, checksum_body: u128) ->
Option<&[u8]>
+where
+ H: iggy_binary_protocol::ConsensusHeader,
+{
+ let body = control_suffix_body(msg);
+ if body.is_empty() {
+ // Nothing to verify. `checksum_body` is irrelevant either way.
+ return Some(body);
+ }
+ if control_body_checksum(body) == checksum_body {
+ Some(body)
+ } else {
+ None
+ }
+}
+
+/// Whether a repaired prepare at `op` falls inside the range this replica is
+/// currently repairing.
+///
+/// A parked log means two things depending on who parked it, and only one is a
+/// repair window. The primary-elect parked the log its merge decided and
repairs
+/// toward exactly that range, so the range IS its scope, including ops at or
+/// below `commit_min`: those are the headers inherited from senders behind the
+/// canonical `log_view`, which the ordinary rule would reject and header
repair
+/// cannot walk back to. A backup's parked `StartView` suffix is only what its
+/// ingest verifies bodies against, and its repair runs for the whole view, so
+/// reading that range as a scope would discard every later op.
+fn repair_op_in_scope(
+ pending: Option<&MergedLog>,
+ is_primary_elect: bool,
+ commit_min: u64,
+ op: u64,
+) -> bool {
+ pending
+ .filter(|_| is_primary_elect)
+ .map_or(op > commit_min, |pending| {
+ (op >= pending.commit_max.max(1) && op <= pending.op_head)
+ || pending
+ .committed_elsewhere
+ .iter()
+ .any(|expected| expected.op == op)
+ })
+}
+
+/// Ceiling on the op range a repair request may ask this replica to walk.
+///
+/// Not `commit_max` alone: a new primary repairing toward a merged log needs
the
+/// uncommitted suffix the view change kept, which sits above every commit
point.
+///
+/// Bounded by the local frontier all the same.
`RequestPreparesHeader::validate`
+/// accepts any `from_op <= to_op`, so `u64::MAX` is legal, and the metadata
serve
+/// path then walks op by op with no `.await` -- on a single-threaded shard
pump
+/// that ends the shard rather than merely serving slowly. Nothing above the
+/// frontier is servable, so the clamp costs nothing.
+fn repair_serve_ceiling(requested_to_op: u64, commit_max: u64, head: u64) ->
u64 {
+ requested_to_op.min(commit_max.max(head))
+}
+
+/// Read this replica's uncommitted suffix out of the metadata journal, for the
+/// window `commit..=op`.
+///
+/// The nack bit is load-bearing, and is set only where absence *proves* this
+/// replica never prepared the op:
+/// * Above the commit point, a missing header is proof: the WAL refuses to
boot
+/// on interior corruption, so a hole in a journal that opened never arrived.
+/// * At or below it, a checkpoint may have compacted the header away. Those
slots
+/// go out blank and un-nacked, read as "no information" rather than licence
to
+/// truncate an op this replica considers committed.
+///
+/// Deriving the suffix on demand is also why it needs no durable record: the
+/// merged log is in memory and bodies are fetched whole, so the WAL is the
only
+/// thing that ever backs a nack and recomputing after a restart gives the same
+/// answer. A torn tail is the one exception, and it changes the answer
correctly:
+/// recovery truncates the incomplete append, which fsyncs before the ack, so
no
+/// replication quorum could have counted it.
+fn build_metadata_dvc_suffix<J>(
+ journal: Option<&J>,
+ commit: u64,
+ op: u64,
+ view_headers: Option<&[PrepareHeader]>,
+) -> DvcSuffix
+where
+ J: JournalHandle,
+ <J as JournalHandle>::Target: Journal<
+ <J as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+{
+ let Some(journal) = journal else {
+ return DvcSuffix::empty();
+ };
+ let handle = journal.handle();
+ build_dvc_suffix(
+ commit,
+ op,
+ |entry_op| {
+ usize::try_from(entry_op)
+ .ok()
+ .and_then(|slot| handle.header(slot))
+ .map(|header| *header)
+ },
+ view_headers,
+ )
+}
+
+/// Plane-independent core of the suffix read. `header_at` answers "do I hold
+/// this op, and what is its header".
+fn build_dvc_suffix(
+ commit: u64,
+ op: u64,
+ header_at: impl Fn(u64) -> Option<PrepareHeader>,
+ view_headers: Option<&[PrepareHeader]>,
+) -> DvcSuffix {
+ // Stitch the adopted view's headers over the journal, high-to-low.
+ //
+ // Reading the journal alone is only correct for a replica whose journal
IS its
+ // log. A backup that adopted a `StartView` is header-poor by design: the
suffix
+ // went to `pending_view_log` and the bodies are still being repaired, so
the
+ // journal holds nothing at those ops and would report them blank AND
nacked,
+ // since a hole above the commit point is normally proof the op never
arrived.
+ // Here it proves only unfinished repair, and enough such senders reach a
nack
+ // quorum against ops the view just decided to keep.
+ //
+ // The head rises to the view's head too, so a later view change cannot
let the
+ // op backtrack below what this replica already acknowledged.
+ let view_head = view_headers
+ .and_then(<[PrepareHeader]>::first)
+ .map_or(0, |header| header.op);
+ let op = op.max(view_head);
+ if op == 0 {
+ return DvcSuffix::empty();
+ }
+ // Window runs from the commit point up, floored at 1 because ops are
1-based.
+ // That floor is a scan bound only: the lines below can raise it above the
+ // commit point, so no reader may read it back as one. See
`merge_commit_max`.
+ let mut low = commit.max(1);
+ if low > op {
+ return DvcSuffix::empty();
+ }
+ if op - low + 1 > DVC_HEADERS_MAX as u64 {
+ // Defensive: every plane's `prepare_queue_depth` is capped below
+ // `DVC_HEADERS_MAX` so `op - commit` cannot reach this. If it does,
the
+ // clamped-away ops go out described by nobody and the merge stalls
rather
+ // than deciding wrongly. Keep the highest entries, whose fate the view
+ // change decides, and log it rather than shipping a different window.
+ let clamped = op - DVC_HEADERS_MAX as u64 + 1;
+ tracing::warn!(
+ commit,
+ op,
+ window_from = clamped,
+ "uncommitted suffix wider than {DVC_HEADERS_MAX} entries;
truncating the DVC window \
+ from below. Ops {}..={} are now undecidable and will stall the
view change",
+ commit + 1,
+ clamped - 1
+ );
+ low = clamped;
+ }
+
+ let len = usize::try_from(op - low + 1).unwrap_or(DVC_HEADERS_MAX);
+ let mut headers = Vec::with_capacity(len);
+ 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) {
+ 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.
+ present_bitset |= 1u128 << index;
+ } else if let Some(header) =
+ view_headers.and_then(|headers| view_header_at(headers, entry_op))
+ {
+ // Held from the adopted view rather than from the journal, so the
+ // header is reported and the op is NOT nacked: this replica knows
+ // the op exists and simply cannot serve its body yet. No present
+ // bit for the same reason.
+ headers.push(*header);
+ } else {
+ headers.push(dvc_blank(entry_op));
+ if entry_op > commit {
Review Comment:
`build_dvc_suffix` pushes a blank for every op it cannot resolve, but only
nacks it when `entry_op > commit`. At `entry_op == commit` the entry goes out
blank *and* un-nacked.
The merge scans inclusive of `commit_max` ("the commit point is scanned so
the adopted log is anchored on a servable header"), so that entry is exactly
the one it needs a verdict on. It gets neither a canonical header nor a nack,
and falls into the `verdict.canonical.is_none() || verdict.copies == 0` branch
in `dvc_merge.rs`:
```
2 of 3 senders blank at the commit point -> AwaitingRepair { undecided_op: 5
}
3 of 3 senders blank at the commit point -> Deadlocked { undecided_op: 5 }
```
Measured against this branch with a scratch test recording `dvc(n, 1, 5, 5,
DvcSuffix::new(vec![dvc_blank(5)], 0, 0))` for each sender.
`Deadlocked` is terminal, so once every replica has reported, the view never
starts.
This is not the `op_head == commit_max` case in general: with a real suffix
the commit-point header is present, and the merge returns `Ready`. It needs
`header_at(commit)` to actually miss.
Whether that is reachable comes down to two things: can journal trimming
advance the snapshot watermark to `commit_max` itself, and can a partition
plane recover a non-zero `commit_max` with an empty in-memory journal? If the
commit point is structurally pinned, the invariant is worth an assertion and a
test, since nothing here enforces it. If it is not pinned, this un-nacked blank
wedges the 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]
