krishvishal commented on code in PR #3824:
URL: https://github.com/apache/iggy/pull/3824#discussion_r3727730228
##########
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:
Not pinned, both planes drop it, and the partition case doesn't even need a
restart.
**Metadata:** not the watermark, the drain. `SnapshotCoordinator::drain` was
`drain(0..=snap_op)` with `snap_op = commit_min`, so the commit point was
removed
outright. Checkpoints fire on journal occupancy, which is symmetric across
replicas.
**Partitions:** `commit_messages` reads `committed_prefix(commit_max)`, it
breaks on
`header.op > commit_max`, inclusive — and `evict_prefix` clears those
headers. A
flushed replica has no resident header at its commit point with the
process still
running. The entry is still servable from the evicted ring; only the
header lookup
missed it, so the present bit was wrong too.
A lagging sender saves the view (its floor op is above its own compaction
point).
The wedge needs the survivors to agree on the commit point.
--
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]
