hubcio commented on code in PR #3824:
URL: https://github.com/apache/iggy/pull/3824#discussion_r3734184277
##########
core/journal/src/lib.rs:
##########
@@ -46,6 +46,28 @@ where
None
}
+ /// Remove every entry at or above `from_op`, returning how many went, and
+ /// leave the snapshot watermark where it is.
+ ///
+ /// Not `drain` with a different range: `drain` advances the watermark
past what
+ /// it removed, which would mark the removed ops evictable when a suffix
+ /// truncation needs them refillable.
+ ///
+ /// Defaults to `Unsupported` rather than a silent zero: "removed nothing"
and
+ /// "cannot remove anything" demand different responses from the caller.
+ ///
+ /// # Errors
+ /// I/O error if the rewrite fails, or `Unsupported` if the implementation
cannot
+ /// truncate.
+ fn truncate_from(&self, _from_op: u64) -> impl Future<Output =
io::Result<usize>> {
Review Comment:
the `Unsupported` default hides missing impls. `JournalState` hand-forwards
every other method - its own doc says to forward EVERY method because a wrapper
that silently substitutes a default for its inner journal is a trap - but
`truncate_from` isn't in its list, and `PartitionJournal` doesn't implement it
either. dropping the default turns both into compile errors instead of a
runtime `Unsupported` in the middle of a view change. two one-line impls, and
the crate isn't published.
##########
core/shard/src/lib.rs:
##########
@@ -3476,7 +3515,16 @@ where
return;
};
let consensus = partition.consensus();
- let actions = consensus.handle_start_view(PlaneKind::Partitions,
&header);
+ let Some(suffix_body) = control_suffix_body_verified(&msg,
header.checksum_body) else {
+ tracing::warn!(
+ shard = self.id,
+ from_replica = header.replica,
+ view = header.view,
+ "dropping start_view whose body failed its checksum"
+ );
+ return;
+ };
+ let actions = consensus.handle_start_view(PlaneKind::Partitions,
&header, suffix_body);
Review Comment:
partition StartView adoption has no divergence reconciliation (the metadata
arm calls it right after adopting), and `PartitionJournal` doesn't implement
`truncate_from`, so it inherits the `Unsupported` default. the failure mode
here is quieter and worse than the metadata plane's wedge: partition `append`
has no slot-collision check - a re-prepared op pushes a duplicate header and
rewrites `op_to_storage_offset` - and `committed_prefix` walks headers
positionally, so the *stale* pre-divergence entry is what `evict_prefix`
flushes to the segment. durable divergent bytes, no error anywhere. `clear_all`
already exists for the state-transfer install; StartView adoption needs the
equivalent.
##########
core/configs/src/server_ng_config/metadata.rs:
##########
@@ -60,10 +60,15 @@ pub const DEFAULT_METADATA_JOURNAL_SLOTS: usize = 1024;
/// margin is `max(this, prepare_queue_depth)`.
pub const METADATA_CHECKPOINT_MARGIN_FLOOR: usize = 64;
-/// Upper bound on `prepare_queue_depth`. Every queued prepare pins a
-/// full message buffer; four thousand in-flight metadata ops is far past
-/// any sane deployment and a likely unit typo.
-pub const MAX_METADATA_PREPARE_QUEUE_DEPTH: usize = 4096;
+/// Upper bound on `prepare_queue_depth`.
+///
+/// Pinned by the view-change wire format, not by memory: a `DoViewChange`
carries
+/// the sender's uncommitted suffix plus one nack bit and one present bit per
entry,
+/// each bitset a single `u128` (`consensus::DVC_HEADERS_MAX` = 128). The
suffix
+/// spans `commit_max..=op`, which this depth bounds, so a deeper queue
produces
+/// entries the new primary can neither adopt nor prove dead. The reserved
head slot
+/// leaves room for the head op.
+pub const MAX_METADATA_PREPARE_QUEUE_DEPTH: usize = 127;
Review Comment:
the drop to 127 is right and load-bearing - `merge_commit_max` treats
`dvc.op - ceiling` as proof of commit, so a peer configured deeper would
manufacture a commit point it never reached. don't relax this under operator
pressure later. but a deployment tuned above 127 now refuses to boot with
"exceeds the maximum (127)" and no hint why; worth naming the cause in the
validation message plus a release note. defaults are 32, so nobody hits it out
of the box.
##########
core/shard/src/lib.rs:
##########
@@ -4322,137 +4434,517 @@ where
h.op = op;
h.namespace = namespace;
h.size = size_of::<RepairRangeReplyHeader>() as u32;
+ h.seal();
});
let _ = self
.bus
.send_to_replica(target, msg.into_generic().into_frozen())
.await;
}
- /// Start metadata tail journal-repair from `peer` when the commit walk
- /// gap-stopped below the known frontier. Shared by `StartView` adoption
- /// and the post-install step of a state transfer.
+ /// Partition-plane twin of [`Self::advance_pending_metadata_view`].
+ ///
+ /// No `RequestPrepares` stream to arm: the partition journal is not
durable
+ /// yet, so coverage either holds or a peer must retransmit. Same invariant
+ /// either way: the view does not start until this replica can serve its
log.
#[allow(clippy::future_not_send)]
- async fn maybe_request_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>, peer: u8)
+ async fn advance_pending_partition_view(&self, namespace: IggyNamespace)
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.is_normal()
- && !consensus.is_transferring()
- && consensus.commit_min() < consensus.commit_max()
- && self.metadata_repair.borrow().is_none()
- {
- let nonce = iggy_common::random_id::get_uuid();
- let to_op = consensus.commit_max();
- let from_op = consensus.commit_min() + 1;
- *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
- nonce,
- to_op,
- peer,
- idle_ticks: 0,
- });
- tracing::info!(
- shard = self.id,
- from_op,
- to_op,
- "metadata behind the group frontier; requesting repair"
- );
- self.send_request_prepares(
- consensus.cluster(),
- consensus.replica(),
- peer,
- nonce,
- from_op,
- to_op,
- consensus.namespace(),
- )
- .await;
+ let partitions = self.plane.partitions();
+ let started = {
+ let Some(partition) = partitions.get_by_ns(&namespace) else {
+ return;
+ };
+ let consensus = partition.consensus();
+ if !consensus.is_primary_for_view(consensus.view()) {
+ return;
+ }
+ let Some(pending) = consensus.pending_view_log() else {
+ return;
+ };
+ let missing = {
+ let journal = partition.log.journal();
+ (pending.commit_max.max(1)..=pending.op_head)
+ .find(|op| journal.inner.header_by_op(*op).is_none())
+ };
+ if let Some(missing_op) = missing {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ missing_op,
+ op_head = pending.op_head,
+ "partition view change waiting on op {missing_op} before
starting the view"
+ );
+ return;
+ }
+
+ let actions = consensus.start_pending_view(PlaneKind::Partitions);
+ let (local_actions, wire_actions) = split_local_actions(actions);
+ // Locals go to the partition dispatcher ONLY: `RebuildPipeline`
+ // executes there (`dispatch_vsr_actions` bails on `journal: None`)
+ // and `CommitJournal` is a no-op in both.
+ dispatch_partition_journal_actions(consensus, partition,
&local_actions).await;
+ // `start_pending_view` flips this replica into `Normal` for the
new
+ // view, so the `StartView` it emits advertises a view the
superblock
+ // must already record. Same gate as the `on_do_view_change` and
+ // `on_start_view` partition arms.
+ if partition.persist_superblock_if_needed().await {
+ dispatch_vsr_actions::<B, _, MJ>(consensus, None,
&wire_actions).await;
+ dispatch_partition_journal_actions(consensus, partition,
&wire_actions).await;
+ }
+ local_actions
+ .iter()
+ .any(|action| matches!(action, VsrAction::CommitJournal))
+ };
+ if started {
+ let config = partitions.config();
+ if let Some(partition) = partitions.get_mut_by_ns(&namespace) {
+ partition.commit_journal(config).await;
+ }
}
}
- #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
- async fn send_request_state_transfer<P>(
- &self,
- consensus: &VsrConsensus<B, P>,
- target: u8,
- nonce: u128,
- ) where
+ /// Re-request the remaining repair window when the stream has gone quiet.
+ ///
+ /// Repair frames are fire-and-forget, so a lost one leaves the session
armed
+ /// forever with the commit walk pinned below the frontier.
+ #[allow(clippy::future_not_send)]
+ async fn retry_stalled_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>)
+ where
B: MessageBus,
P: Pipeline<Entry = consensus::PipelineEntry>,
{
- let msg =
-
Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
- .transmute_header(|_, h: &mut RequestStateTransferHeader| {
- h.command = Command2::RequestStateTransfer;
- h.cluster = consensus.cluster();
- h.replica = consensus.replica();
- h.nonce = nonce;
- h.namespace = consensus.namespace();
- h.size = size_of::<RequestStateTransferHeader>() as u32;
- });
- let _ = self
- .bus
- .send_to_replica(target, msg.into_generic().into_frozen())
- .await;
+ // Stall retry (mirrors `tick_partitions`): a lost frame must not
wedge it.
+ let repair_retry_ticks = self.repair_retry_ticks.get();
+ let stalled = {
+ // `ViewChange` too: a parked view change repairs toward its
merged log
+ // and cannot start until the window fills. Gating on `Normal`
alone
+ // defers a dropped frame to the 500-tick escalation.
+ let repairing_view = consensus.pending_view_log().is_some()
+ && consensus.is_primary_for_view(consensus.view());
+ let mut session = self.metadata_repair.borrow_mut();
+ session.as_mut().and_then(|session| {
+ if !consensus.is_normal() && !repairing_view {
+ return None;
+ }
+ session.idle_ticks += 1;
+ if session.idle_ticks < repair_retry_ticks {
+ return None;
+ }
+ session.idle_ticks = 0;
+ Some((session.peer, session.nonce, session.to_op))
+ })
+ };
+ if let Some((peer, nonce, to_op)) = stalled {
+ // Primary-elect only. Its window starts at the merged log's commit
+ // point, which can sit below local `commit_min` (the headers
inherited
+ // from senders behind the canonical log_view live there), so
+ // `commit_min + 1` would skip them. A backup's parked `StartView`
+ // suffix is only a verification reference; resuming from its
commit
+ // point would restart at the view's opening head, not at the gap.
+ let from_op = consensus
+ .pending_view_log()
+ .filter(|_| consensus.is_primary_for_view(consensus.view()))
+ .map_or_else(
+ || consensus.commit_min() + 1,
+ |pending| pending.commit_max.max(1),
+ );
+ if from_op <= to_op {
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ peer,
+ "metadata repair stalled; re-requesting remaining window"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
}
- /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
- /// `available = 0` (the requester falls back to journal repair or
- /// retries elsewhere); an offer ships its encoded state manifest as the
- /// frame body.
- #[allow(
- clippy::future_not_send,
- clippy::cast_possible_truncation,
- clippy::too_many_arguments
- )]
- async fn send_state_transfer_target(
- &self,
- cluster: u128,
- self_id: u8,
- target: u8,
- nonce: u128,
- namespace: u64,
- descriptor: TransferDescriptor<'_>,
- ) where
+ /// Compare a backup's log against the headers the concluding `StartView`
+ /// published, and report where they disagree.
+ ///
+ /// Without this, divergence is silent and permanent: the backup acks with
its
+ /// own checksum, the primary rejects the ack, and journal repair skips an
op
+ /// it already has a header for.
+ ///
+ /// The split at the announced commit point is what matters. Above it a
+ /// disagreement is ordinary, so the entry is dropped and the primary's
+ /// retransmission refills the range. At or below it, this replica applied
+ /// something the view says was different, which only state transfer
fixes, so
+ /// it is reported and left alone.
+ ///
+ /// Truncation uses `Journal::truncate_from`, not `drain`: `drain` advances
+ /// `snapshot_op` past what it removed, marking ops that must stay
refillable
+ /// as evictable.
+ #[allow(clippy::future_not_send)]
+ async fn reconcile_metadata_view_divergence(&self)
+ where
B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: MetadataStm,
{
- let manifest = descriptor
- .offer
- .map(|(entries, _)| consensus::encode_state_manifest(entries));
- let total_size =
- size_of::<StateTransferTargetHeader>() +
manifest.as_ref().map_or(0, Vec::len);
- let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
- if let Some(manifest) = &manifest {
-
msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ let metadata = self.plane.metadata();
+ let Some(ref consensus) = metadata.consensus else {
+ return;
+ };
+ // Backups only; a primary reconciles through the merge itself.
+ if consensus.is_primary_for_view(consensus.view()) {
+ return;
}
- let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
- h.command = Command2::StateTransferTarget;
- h.cluster = cluster;
- h.replica = self_id;
- h.nonce = nonce;
- h.namespace = namespace;
- h.size = total_size as u32;
- // The serving replica's own progress travels with every
descriptor,
- // available or not: it is what lets a receiver refuse an offer
from
- // a replica that knows less than it does.
- h.view = descriptor.view;
- h.commit_max = descriptor.commit_max;
- h.unavailable_transient = u8::from(descriptor.transient);
- if let Some((_, commit_op)) = descriptor.offer {
- h.available = 1;
- h.commit_op = commit_op;
- }
- });
- let _ = self
- .bus
- .send_to_replica(target, msg.into_generic().into_frozen())
- .await;
- }
+ let Some(pending) = consensus.pending_view_log() else {
+ return;
+ };
+ let Some(journal) = metadata.journal.as_ref() else {
+ return;
+ };
- #[allow(
- clippy::future_not_send,
- clippy::cast_possible_truncation,
+ // Truncation is safe only above what this replica has *applied*,
which is
+ // not the view's commit point: `pending.commit_max` is the new
primary's
+ // number and a backup can sit above it. Splitting on the view's number
+ // would drop already-executed ops with no rollback, and silently.
+ let applied_floor = pending.commit_max.max(consensus.commit_min());
+
+ let mut repairable_from: Option<u64> = None;
+ for canonical in &pending.headers {
+ let Some(local) = usize::try_from(canonical.op)
+ .ok()
+ .and_then(|slot| journal.handle().header(slot))
+ else {
+ continue;
+ };
+ if local.checksum == canonical.checksum {
+ continue;
+ }
+ if canonical.op <= applied_floor {
+ tracing::error!(
+ shard = self.id,
+ op = canonical.op,
+ view = consensus.view(),
+ commit_max = pending.commit_max,
+ commit_min = consensus.commit_min(),
+ local_checksum = local.checksum,
+ canonical_checksum = canonical.checksum,
+ "committed op {} disagrees with the view that just
started; this replica \
+ applied a different op as committed and cannot be
reconciled by log repair",
+ canonical.op
+ );
+ continue;
+ }
+ repairable_from = Some(repairable_from.map_or(canonical.op, |op|
op.min(canonical.op)));
+ }
+
+ let Some(from_op) = repairable_from else {
+ return;
+ };
+ match journal.handle().truncate_from(from_op).await {
+ Ok(removed) => {
+ // The snapshot's `(op, commit)` tag does not move when
entries are
+ // removed under it, so the next `DoViewChange` would
advertise the
+ // dropped headers and offer bodies this replica cannot serve.
+ consensus.invalidate_local_dvc_suffix();
+ tracing::warn!(
+ shard = self.id,
+ from_op,
+ removed,
+ op_head = pending.op_head,
+ view = consensus.view(),
+ "dropped {removed} uncommitted entries from op {from_op}
that disagreed with \
+ the view's log; the primary's retransmission refills the
range"
+ );
+ }
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ from_op,
+ %error,
+ "could not drop the diverging uncommitted entries from op
{from_op}; journal \
+ repair skips ops it already holds a header for, so this
replica will not \
+ converge at those ops until it is restarted"
+ );
+ }
+ }
+ }
+
+ /// Drive a parked view change to completion.
+ ///
+ /// A DVC quorum decides the log before this replica necessarily holds it,
so
+ /// the merged log parks in consensus and this replica stays in
`ViewChange`,
+ /// announcing and preparing nothing: `StartView` promises it can serve
every
+ /// op it names, and a backup adopting that head asks for the bodies at
once.
+ ///
+ /// Check coverage, then start the view or pull missing bodies from a peer
that
+ /// offered them in its DVC. Only those peers: a cleared present bit means
the
+ /// body was never held or cannot be read back.
+ #[allow(clippy::future_not_send)]
+ async fn advance_pending_metadata_view(&self)
+ where
+ B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: MetadataStm,
+ {
+ let metadata = self.plane.metadata();
+ let Some(ref consensus) = metadata.consensus else {
+ return;
+ };
+ // Primary-elect only. A backup's parked `StartView` suffix is only
what
+ // its ingest verifies bodies against; driving repair from it would
put a
+ // rejoining node on the tail-repair path when its gap sits below every
+ // peer's retention floor, racing the view probe that picks state
transfer.
+ if !consensus.is_primary_for_view(consensus.view()) {
+ return;
+ }
+ let Some(pending) = consensus.pending_view_log() else {
+ return;
+ };
+ let Some(journal) = metadata.journal.as_ref() else {
+ return;
+ };
+
+ let held = |op: u64| {
Review Comment:
`held` answers by presence only - `journal.header(slot).is_some()` - never
by identity. a primary-elect below the canonical `log_view` holding a
*different* prepare at an op the merge kept sees `held() == true`, skips
repair, and starts the view: `StartView` ships the merged (canonical) headers
while the pipeline rebuild reads the local journal and self-acks the divergent
entry. a backup that already holds the canonical header skips the append on
retransmit (`on_replicate`'s already-journaled path) and acks the *received*
header's checksum, and `handle_prepare_ok` compares that against the primary's
own pipeline entry - both sides of the gate are the primary's header, so quorum
forms and the divergent op commits while the up-to-date backup applies the
canonical one. two replicas apply different operations at the same committed,
client-acked op, silently.
reachable 2-of-3: the primary-elect contributes exactly one implicit nack,
short of the nack quorum, so the canonical header survives the merge and
nothing ever asks the primary to repair. the backup path below does the
identity compare (`local.checksum == canonical.checksum`) that this closure
omits, and `reconcile_metadata_view_divergence` returns early for the primary.
`held` has to match the journaled header's checksum against
`pending.headers`/`committed_elsewhere`, and a mismatching primary-elect needs
the same truncate + repair a backup gets. the partition twin has the same
presence-only check and doesn't consult `committed_elsewhere` at all.
##########
core/consensus/src/impls.rs:
##########
@@ -2818,6 +3137,110 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
// the loopback queue directly.
self.loopback_queue.borrow_mut().clear();
+ tracing::info!(
+ replica = self.replica,
+ view = self.view.get(),
+ op_head = new_op,
+ commit_max = max_commit,
+ "view-change quorum merged; repairing up to the merged log before
starting the view"
+ );
+ emit_replica_event(
+ SimEventKind::ReplicaStateChanged,
+ &ReplicaLogContext::from_consensus(self, plane),
+ );
+
+ // No sends yet: `SendStartView` promises this replica can serve every
op in
+ // the merged log, and a backup adopting the announced head asks it
for the
+ // bodies behind it.
+ Vec::new()
+ }
+
+ /// Sizes handed to the DVC merge.
+ const fn merge_quorums(&self) -> MergeQuorums {
+ MergeQuorums {
+ view_change: self.quorum_view_change(),
+ nack_prepare: self.quorum_nack_prepare(),
+ replica_count: self.replica_count as usize,
+ // The cluster-wide ceiling, not `self.prepare_queue_max`: this
node's
+ // config says nothing about how deep a peer's pipeline is.
+ prepare_queue_ceiling: PREPARE_QUEUE_CEILING as u64,
+ }
+ }
+
+ /// The merged log this replica is repairing toward, if a view change is
+ /// mid-transition. The shard reads it for the op range it must cover
before the
+ /// view can start, and for which peers offered the bodies.
+ #[must_use]
+ pub fn pending_view_log(&self) -> Option<MergedLog> {
+ self.pending_view_log.borrow().clone()
+ }
+
+ /// Replicas that offered a body for `op`, most-recent-log_view first.
+ ///
+ /// Only meaningful while a merge is parked. These peers and nobody else: a
+ /// cleared present bit means the body was never held or cannot be read
back,
+ /// and the view change is blocked on the round-trip.
+ #[must_use]
+ pub fn pending_view_body_sources(&self, op: u64) -> Vec<u8> {
+ let quorum = self.do_view_change_from_all_replicas.borrow();
+ let mut sources: Vec<(u32, u8)> = dvc_iter(&quorum)
+ .filter(|dvc| dvc.replica != self.replica)
+ .filter_map(|dvc| {
+ let index = dvc.suffix.index_of(dvc.op, op)?;
+ dvc.suffix
+ .offers_body(index)
+ .then_some((dvc.log_view, dvc.replica))
+ })
+ .collect();
+ sources.sort_unstable_by_key(|(log_view, _)|
std::cmp::Reverse(*log_view));
+ sources.into_iter().map(|(_, replica)| replica).collect()
+ }
+
+ /// Finish the parked view change: this replica's journal now covers the
merged
+ /// log, so it can serve any op it is about to announce.
+ ///
+ /// Called by the shard after repair progress. No-op when nothing is
parked.
+ ///
+ /// # Panics
+ /// If the merged uncommitted range exceeds pipeline capacity, which needs
a head
+ /// more than one pipeline depth above the proven commit point.
+ pub fn start_pending_view(&self, plane: PlaneKind) -> Vec<VsrAction> {
Review Comment:
this is the only exit from a view change that doesn't reset the DVC quorum -
the other five paths all go through `reset_view_change_state()`. the winning
primary therefore keeps `do_view_change_from_all_replicas`, now carrying a
suffix `Vec` per sender, for its whole primaryship, per partition it leads.
nothing reads the array after the view starts (a late same-view DVC returns at
the status gate before touching it; a higher-view one resets anyway), so
freeing the suffix storage here (`dvc_reset` + `invalidate_local_dvc_suffix`)
is safe - but not `reset_dvc_quorum`, which also clears the
`do_view_change_quorum` latch that now means "log decided".
##########
core/shard/src/lib.rs:
##########
@@ -7438,10 +7914,411 @@ 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.
+///
+/// Read through `repair_header`, not the resident headers: the committed
prefix
+/// leaves those as soon as its bytes reach a segment, which on a caught-up
+/// replica includes the commit point itself.
+fn refresh_partition_dvc_suffix<B, SB>(partition:
&partitions::IggyPartition<B, SB>)
+where
+ B: MessageBus,
+ SB: SuperblockStore,
+{
+ 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.repair_header(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`
Review Comment:
the doc here already names the weak `validate` and the no-await walk - this
covers the half it doesn't: the ceiling bounds `to_op`, but nothing floors
`from_op`. `RequestPreparesHeader::validate` only requires `1 <= from_op <=
to_op`, and the metadata skip loop in `on_request_prepares` walks op by op from
whatever the peer sent. probes are cheap but the loop is unbounded - `from_op =
1` against a big compacted frontier is a long non-yielding stretch on the pump,
repeatable at will, against a 10 ms consensus tick. the partition arm already
jumps straight to `retained_from`, which is exactly the shape: floor the
metadata side at `snapshot_op() + 1` (and/or bound `to_op - from_op` in
validate). the `to_op` raise to `max(commit_max, head)` is required - the
merged suffix above commit_max has to stay servable - so the fix is the floor
only.
##########
core/server_common/src/send_messages2.rs:
##########
@@ -678,6 +678,24 @@ fn transcode_legacy_request(
/// chunk and steps by `batch_length`. Callers whose buffer is meant to BE the
/// batch must reject the surplus themselves - see [`convert_request_message`].
pub fn decode_batch_slice(body: &[u8]) -> Result<SendMessages2Ref<'_>,
IggyError> {
+ decode_batch_slice_verified(body, true)
+}
+
+/// [`decode_batch_slice`] with the checksum check made optional.
+///
+/// `verify_checksum` exists for the disk-poll path, whose operator knob
decides
+/// whether a read pays for a full re-hash of every batch. The layout checks
are not
+/// optional either way: a short or self-inconsistent record is rejected
regardless,
+/// because the caller would otherwise index past it.
+///
+/// # Errors
+/// [`IggyError::InvalidCommand`] for a short or inconsistent record, and
+/// [`IggyError::InvalidBatchChecksum`] when verification is on and the batch
does not
+/// match. Callers that must tell corruption from a partial tail need both.
+pub fn decode_batch_slice_verified(
Review Comment:
the name promises "verified" but the second argument turns verification off
- poll_plan passes the operator knob straight through, so the common disk-poll
read runs `..._verified(.., false)`. rename, or take a two-variant enum instead
of the bool so the call site reads honestly.
##########
core/shard/src/lib.rs:
##########
@@ -4322,137 +4434,517 @@ where
h.op = op;
h.namespace = namespace;
h.size = size_of::<RepairRangeReplyHeader>() as u32;
+ h.seal();
});
let _ = self
.bus
.send_to_replica(target, msg.into_generic().into_frozen())
.await;
}
- /// Start metadata tail journal-repair from `peer` when the commit walk
- /// gap-stopped below the known frontier. Shared by `StartView` adoption
- /// and the post-install step of a state transfer.
+ /// Partition-plane twin of [`Self::advance_pending_metadata_view`].
+ ///
+ /// No `RequestPrepares` stream to arm: the partition journal is not
durable
+ /// yet, so coverage either holds or a peer must retransmit. Same invariant
+ /// either way: the view does not start until this replica can serve its
log.
#[allow(clippy::future_not_send)]
- async fn maybe_request_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>, peer: u8)
+ async fn advance_pending_partition_view(&self, namespace: IggyNamespace)
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.is_normal()
- && !consensus.is_transferring()
- && consensus.commit_min() < consensus.commit_max()
- && self.metadata_repair.borrow().is_none()
- {
- let nonce = iggy_common::random_id::get_uuid();
- let to_op = consensus.commit_max();
- let from_op = consensus.commit_min() + 1;
- *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
- nonce,
- to_op,
- peer,
- idle_ticks: 0,
- });
- tracing::info!(
- shard = self.id,
- from_op,
- to_op,
- "metadata behind the group frontier; requesting repair"
- );
- self.send_request_prepares(
- consensus.cluster(),
- consensus.replica(),
- peer,
- nonce,
- from_op,
- to_op,
- consensus.namespace(),
- )
- .await;
+ let partitions = self.plane.partitions();
+ let started = {
+ let Some(partition) = partitions.get_by_ns(&namespace) else {
+ return;
+ };
+ let consensus = partition.consensus();
+ if !consensus.is_primary_for_view(consensus.view()) {
+ return;
+ }
+ let Some(pending) = consensus.pending_view_log() else {
+ return;
+ };
+ let missing = {
+ let journal = partition.log.journal();
+ (pending.commit_max.max(1)..=pending.op_head)
+ .find(|op| journal.inner.header_by_op(*op).is_none())
+ };
+ if let Some(missing_op) = missing {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ missing_op,
+ op_head = pending.op_head,
+ "partition view change waiting on op {missing_op} before
starting the view"
+ );
+ return;
+ }
+
+ let actions = consensus.start_pending_view(PlaneKind::Partitions);
+ let (local_actions, wire_actions) = split_local_actions(actions);
+ // Locals go to the partition dispatcher ONLY: `RebuildPipeline`
+ // executes there (`dispatch_vsr_actions` bails on `journal: None`)
+ // and `CommitJournal` is a no-op in both.
+ dispatch_partition_journal_actions(consensus, partition,
&local_actions).await;
+ // `start_pending_view` flips this replica into `Normal` for the
new
+ // view, so the `StartView` it emits advertises a view the
superblock
+ // must already record. Same gate as the `on_do_view_change` and
+ // `on_start_view` partition arms.
+ if partition.persist_superblock_if_needed().await {
+ dispatch_vsr_actions::<B, _, MJ>(consensus, None,
&wire_actions).await;
+ dispatch_partition_journal_actions(consensus, partition,
&wire_actions).await;
+ }
+ local_actions
+ .iter()
+ .any(|action| matches!(action, VsrAction::CommitJournal))
+ };
+ if started {
+ let config = partitions.config();
+ if let Some(partition) = partitions.get_mut_by_ns(&namespace) {
+ partition.commit_journal(config).await;
+ }
}
}
- #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
- async fn send_request_state_transfer<P>(
- &self,
- consensus: &VsrConsensus<B, P>,
- target: u8,
- nonce: u128,
- ) where
+ /// Re-request the remaining repair window when the stream has gone quiet.
+ ///
+ /// Repair frames are fire-and-forget, so a lost one leaves the session
armed
+ /// forever with the commit walk pinned below the frontier.
+ #[allow(clippy::future_not_send)]
+ async fn retry_stalled_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>)
+ where
B: MessageBus,
P: Pipeline<Entry = consensus::PipelineEntry>,
{
- let msg =
-
Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
- .transmute_header(|_, h: &mut RequestStateTransferHeader| {
- h.command = Command2::RequestStateTransfer;
- h.cluster = consensus.cluster();
- h.replica = consensus.replica();
- h.nonce = nonce;
- h.namespace = consensus.namespace();
- h.size = size_of::<RequestStateTransferHeader>() as u32;
- });
- let _ = self
- .bus
- .send_to_replica(target, msg.into_generic().into_frozen())
- .await;
+ // Stall retry (mirrors `tick_partitions`): a lost frame must not
wedge it.
+ let repair_retry_ticks = self.repair_retry_ticks.get();
+ let stalled = {
+ // `ViewChange` too: a parked view change repairs toward its
merged log
+ // and cannot start until the window fills. Gating on `Normal`
alone
+ // defers a dropped frame to the 500-tick escalation.
+ let repairing_view = consensus.pending_view_log().is_some()
+ && consensus.is_primary_for_view(consensus.view());
+ let mut session = self.metadata_repair.borrow_mut();
+ session.as_mut().and_then(|session| {
+ if !consensus.is_normal() && !repairing_view {
+ return None;
+ }
+ session.idle_ticks += 1;
+ if session.idle_ticks < repair_retry_ticks {
+ return None;
+ }
+ session.idle_ticks = 0;
+ Some((session.peer, session.nonce, session.to_op))
+ })
+ };
+ if let Some((peer, nonce, to_op)) = stalled {
+ // Primary-elect only. Its window starts at the merged log's commit
+ // point, which can sit below local `commit_min` (the headers
inherited
+ // from senders behind the canonical log_view live there), so
+ // `commit_min + 1` would skip them. A backup's parked `StartView`
+ // suffix is only a verification reference; resuming from its
commit
+ // point would restart at the view's opening head, not at the gap.
+ let from_op = consensus
+ .pending_view_log()
+ .filter(|_| consensus.is_primary_for_view(consensus.view()))
+ .map_or_else(
+ || consensus.commit_min() + 1,
+ |pending| pending.commit_max.max(1),
+ );
+ if from_op <= to_op {
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ peer,
+ "metadata repair stalled; re-requesting remaining window"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
}
- /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
- /// `available = 0` (the requester falls back to journal repair or
- /// retries elsewhere); an offer ships its encoded state manifest as the
- /// frame body.
- #[allow(
- clippy::future_not_send,
- clippy::cast_possible_truncation,
- clippy::too_many_arguments
- )]
- async fn send_state_transfer_target(
- &self,
- cluster: u128,
- self_id: u8,
- target: u8,
- nonce: u128,
- namespace: u64,
- descriptor: TransferDescriptor<'_>,
- ) where
+ /// Compare a backup's log against the headers the concluding `StartView`
+ /// published, and report where they disagree.
+ ///
+ /// Without this, divergence is silent and permanent: the backup acks with
its
+ /// own checksum, the primary rejects the ack, and journal repair skips an
op
+ /// it already has a header for.
+ ///
+ /// The split at the announced commit point is what matters. Above it a
+ /// disagreement is ordinary, so the entry is dropped and the primary's
+ /// retransmission refills the range. At or below it, this replica applied
+ /// something the view says was different, which only state transfer
fixes, so
+ /// it is reported and left alone.
+ ///
+ /// Truncation uses `Journal::truncate_from`, not `drain`: `drain` advances
+ /// `snapshot_op` past what it removed, marking ops that must stay
refillable
+ /// as evictable.
+ #[allow(clippy::future_not_send)]
+ async fn reconcile_metadata_view_divergence(&self)
+ where
B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: MetadataStm,
{
- let manifest = descriptor
- .offer
- .map(|(entries, _)| consensus::encode_state_manifest(entries));
- let total_size =
- size_of::<StateTransferTargetHeader>() +
manifest.as_ref().map_or(0, Vec::len);
- let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
- if let Some(manifest) = &manifest {
-
msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ let metadata = self.plane.metadata();
+ let Some(ref consensus) = metadata.consensus else {
+ return;
+ };
+ // Backups only; a primary reconciles through the merge itself.
+ if consensus.is_primary_for_view(consensus.view()) {
+ return;
}
- let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
- h.command = Command2::StateTransferTarget;
- h.cluster = cluster;
- h.replica = self_id;
- h.nonce = nonce;
- h.namespace = namespace;
- h.size = total_size as u32;
- // The serving replica's own progress travels with every
descriptor,
- // available or not: it is what lets a receiver refuse an offer
from
- // a replica that knows less than it does.
- h.view = descriptor.view;
- h.commit_max = descriptor.commit_max;
- h.unavailable_transient = u8::from(descriptor.transient);
- if let Some((_, commit_op)) = descriptor.offer {
- h.available = 1;
- h.commit_op = commit_op;
- }
- });
- let _ = self
- .bus
- .send_to_replica(target, msg.into_generic().into_frozen())
- .await;
- }
+ let Some(pending) = consensus.pending_view_log() else {
+ return;
+ };
+ let Some(journal) = metadata.journal.as_ref() else {
+ return;
+ };
- #[allow(
- clippy::future_not_send,
- clippy::cast_possible_truncation,
+ // Truncation is safe only above what this replica has *applied*,
which is
+ // not the view's commit point: `pending.commit_max` is the new
primary's
+ // number and a backup can sit above it. Splitting on the view's number
+ // would drop already-executed ops with no rollback, and silently.
+ let applied_floor = pending.commit_max.max(consensus.commit_min());
+
+ let mut repairable_from: Option<u64> = None;
+ for canonical in &pending.headers {
+ let Some(local) = usize::try_from(canonical.op)
+ .ok()
+ .and_then(|slot| journal.handle().header(slot))
+ else {
+ continue;
+ };
+ if local.checksum == canonical.checksum {
+ continue;
+ }
+ if canonical.op <= applied_floor {
+ tracing::error!(
+ shard = self.id,
+ op = canonical.op,
+ view = consensus.view(),
+ commit_max = pending.commit_max,
+ commit_min = consensus.commit_min(),
+ local_checksum = local.checksum,
+ canonical_checksum = canonical.checksum,
+ "committed op {} disagrees with the view that just
started; this replica \
+ applied a different op as committed and cannot be
reconciled by log repair",
+ canonical.op
+ );
+ continue;
+ }
+ repairable_from = Some(repairable_from.map_or(canonical.op, |op|
op.min(canonical.op)));
+ }
+
+ let Some(from_op) = repairable_from else {
+ return;
+ };
+ match journal.handle().truncate_from(from_op).await {
Review Comment:
`truncate_from` shares the drain guard, so it's excluded against `drain` and
other rewrites - but nothing excludes it against an append. that exclusion is
metadata's `journal_gate`, a private field this crate can't reach, and
`on_repair_prepare`'s append is in the same position. it appears to hold today
by call-site placement (backups-only caller, submits gated on primaryship),
which nothing documents or asserts - and detached submit tasks are explicitly
named as concurrent drivers of the gated section, so the serialization isn't
self-evident. worth routing both shard-side journal mutations through
metadata-crate methods that take the gate, or at minimum a comment here
recording why no append can interleave - the reasoning isn't visible at the
call site.
##########
core/consensus/src/impls.rs:
##########
@@ -3062,15 +3485,25 @@ where
// sealed region before the entry is journaled. Leaving those prepares
at `0`
// is the designed "nothing to verify" sentinel, so a future durable
partition
// journal skips verification instead of failing every entry as
corrupt.
+ //
+ // So a partition prepare's `checksum` identifies its header alone,
and two
+ // such prepares at one op with matching header fields are
indistinguishable
+ // to the view-change merge. Closing that wants `checksum_body` here
to BE
+ // the batch checksum, recomputed after stamping, so `checksum` covers
the
+ // body for free by hashing this field.
+ //
+ // Bounded by `size`, the range every verifier re-reads; the prepare
+ // inherits it verbatim below.
let checksum_body = if consensus.namespace ==
METADATA_CONSENSUS_NAMESPACE {
Review Comment:
partition prepares still seal `checksum_body = 0`, so `checksum` covers the
header alone and the batch bytes ride unverified. the comment above already
says it, but the merge consequence is worth spelling out: the dvc merge counts
a replica whose batch *bytes* differ as a servable copy of the canonical
header, and partition repair ingest has no merged-log identity gate, so the
body branch of `verify_prepare_integrity` short-circuits on the zero and a
divergent batch passes on the serving peer's word. the fix the comment names
(make this the batch checksum, recomputed after stamping) closes it - "sealed
on both planes" in the PR text is true of the field, not yet of the coverage.
##########
core/partitions/src/poll_plan.rs:
##########
@@ -533,14 +536,23 @@ impl DiskReadPlan {
faulted = true;
break 'walk;
};
- let consumed = walk_disk_chunk(
+ let ChunkWalk { consumed, corrupt } = walk_disk_chunk(
&chunk,
query,
count,
&mut matched,
&mut fragments,
&mut last_matching_offset,
+ self.validate_checksum,
Review Comment:
`validate_checksum` ships `false`, and that flips the disk poll from
fail-closed to fail-open. on master this path went through
`decode_batch_slice`, which verified `batch_checksum` unconditionally and never
served a failing batch; with the shipped default only the layout check runs and
a corrupt batch is decoded and handed to the consumer.
`read_disk_faults_closed_on_batch_checksum_mismatch` even asserts `Matched`
with the knob off - a test asserting corrupt data is served. the PR text says
"disk polls verify each batch against its `batch_checksum` and fail closed",
which isn't what ships. flipping the default restores master's behavior and
master's cost - the hash is on the order of the memcpy the poll already
performs.
three adjacent things. `Faulted` maps to an empty poll with no error
surfaced, and at-rest corruption never "reads again", so with the knob *on* the
consumer stalls forever, indistinguishable from caught-up - wants an error to
the client, independently of the default. the comment above the knob in
server-ng's config.toml still calls it inert and claims boot warns when set
(that warn got deleted). and the same key now means a one-time boot sweep on
the legacy server vs a per-poll hash forever on ng - operator-facing text
should say so.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -98,17 +178,43 @@ async fn read_persisted_offset(path: &str) ->
Result<Option<u64>, IggyError> {
.open(path)
.await
.map_err(|_|
IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
- let buf = vec![0u8; OFFSET_SIZE];
- let compio::BufResult(read, buf) = file.read_exact_at(buf, 0).await;
- match read {
- Ok(()) => {}
- Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof =>
return Ok(None),
+ // Read the whole record, falling back to a bare offset: a pre-checksum
file is
+ // exactly `OFFSET_SIZE` long, so the first read reports EOF rather than
failing.
+ let compio::BufResult(read, buf) = file.read_exact_at(vec![0u8;
OFFSET_RECORD_SIZE], 0).await;
+ let bytes = match read {
+ Ok(()) => buf,
+ Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
+ let compio::BufResult(read, legacy) =
+ file.read_exact_at(vec![0u8; OFFSET_SIZE], 0).await;
+ match read {
+ Ok(()) => legacy,
+ Err(error) if error.kind() ==
std::io::ErrorKind::UnexpectedEof => {
+ return Ok(None);
+ }
+ Err(_) => return
Err(IggyError::CannotReadConsumerOffsets(path.to_owned())),
+ }
+ }
Err(_) => return
Err(IggyError::CannotReadConsumerOffsets(path.to_owned())),
+ };
+ match decode_offset_record(&bytes) {
+ OffsetRecord::Value { offset, .. } => Ok(Some(offset)),
+ OffsetRecord::Torn => Ok(None),
+ OffsetRecord::Corrupt {
+ offset,
+ expected,
+ found,
+ } => {
+ tracing::error!(
+ path,
+ offset,
+ expected,
+ found,
+ "consumer offset file failed its checksum; refusing to fold a
value that may \
+ rewind or skip the consumer"
+ );
+ Err(IggyError::CannotReadConsumerOffsets(path.to_owned()))
Review Comment:
this `Err` ends in a shard panic. `persist_offset_max` propagates it, the
commit walk turns the failed consumer-offset commit into the "replica is
divergent from cluster commit; restart required" `panic!`, and the boot loader
skips a corrupt file *without unlinking it* - while the in-memory offset map
starts empty every boot, so the first auto-commit touching that key re-reads
the file and panics again. one corrupt offset file is a persistent landmine:
crash, restart, crash. only the auto-commit read path is affected (an explicit
store blind-writes and repairs the file).
the two readers should agree: boot already treats corrupt as absent and
keeps going. do the same here - log it, fold from `max_committed` alone, and
never route a data-file checksum failure into the commit-walk fatal path.
##########
core/binary_protocol/src/consensus/error.rs:
##########
@@ -29,6 +29,15 @@ pub enum ConsensusError {
#[error("invalid checksum")]
InvalidChecksum,
+ #[error(
+ "{command:?}: header checksum {found:#034x} does not cover the frame
(expected {expected:#034x})"
Review Comment:
a pre-seal peer's frame surfaces with `found` printing as 0x000...0, and the
drop gets logged as "dropping message with invalid command" (router and shard
both) even though the command parsed fine. mixed-version is a declared hard
break, so the operator diagnosing it deserves better: retitle the drop log, and
when `found == 0` add a hint that this looks like a peer predating the frame
seal - that's the signature every pre-seal node sends.
##########
core/shard/src/lib.rs:
##########
@@ -4322,137 +4434,517 @@ where
h.op = op;
h.namespace = namespace;
h.size = size_of::<RepairRangeReplyHeader>() as u32;
+ h.seal();
});
let _ = self
.bus
.send_to_replica(target, msg.into_generic().into_frozen())
.await;
}
- /// Start metadata tail journal-repair from `peer` when the commit walk
- /// gap-stopped below the known frontier. Shared by `StartView` adoption
- /// and the post-install step of a state transfer.
+ /// Partition-plane twin of [`Self::advance_pending_metadata_view`].
+ ///
+ /// No `RequestPrepares` stream to arm: the partition journal is not
durable
+ /// yet, so coverage either holds or a peer must retransmit. Same invariant
+ /// either way: the view does not start until this replica can serve its
log.
#[allow(clippy::future_not_send)]
- async fn maybe_request_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>, peer: u8)
+ async fn advance_pending_partition_view(&self, namespace: IggyNamespace)
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.is_normal()
- && !consensus.is_transferring()
- && consensus.commit_min() < consensus.commit_max()
- && self.metadata_repair.borrow().is_none()
- {
- let nonce = iggy_common::random_id::get_uuid();
- let to_op = consensus.commit_max();
- let from_op = consensus.commit_min() + 1;
- *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
- nonce,
- to_op,
- peer,
- idle_ticks: 0,
- });
- tracing::info!(
- shard = self.id,
- from_op,
- to_op,
- "metadata behind the group frontier; requesting repair"
- );
- self.send_request_prepares(
- consensus.cluster(),
- consensus.replica(),
- peer,
- nonce,
- from_op,
- to_op,
- consensus.namespace(),
- )
- .await;
+ let partitions = self.plane.partitions();
+ let started = {
+ let Some(partition) = partitions.get_by_ns(&namespace) else {
+ return;
+ };
+ let consensus = partition.consensus();
+ if !consensus.is_primary_for_view(consensus.view()) {
+ return;
+ }
+ let Some(pending) = consensus.pending_view_log() else {
+ return;
+ };
+ let missing = {
+ let journal = partition.log.journal();
+ (pending.commit_max.max(1)..=pending.op_head)
+ .find(|op| journal.inner.header_by_op(*op).is_none())
+ };
+ if let Some(missing_op) = missing {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ missing_op,
+ op_head = pending.op_head,
+ "partition view change waiting on op {missing_op} before
starting the view"
+ );
+ return;
+ }
+
+ let actions = consensus.start_pending_view(PlaneKind::Partitions);
+ let (local_actions, wire_actions) = split_local_actions(actions);
+ // Locals go to the partition dispatcher ONLY: `RebuildPipeline`
+ // executes there (`dispatch_vsr_actions` bails on `journal: None`)
+ // and `CommitJournal` is a no-op in both.
+ dispatch_partition_journal_actions(consensus, partition,
&local_actions).await;
+ // `start_pending_view` flips this replica into `Normal` for the
new
+ // view, so the `StartView` it emits advertises a view the
superblock
+ // must already record. Same gate as the `on_do_view_change` and
+ // `on_start_view` partition arms.
+ if partition.persist_superblock_if_needed().await {
+ dispatch_vsr_actions::<B, _, MJ>(consensus, None,
&wire_actions).await;
+ dispatch_partition_journal_actions(consensus, partition,
&wire_actions).await;
+ }
+ local_actions
+ .iter()
+ .any(|action| matches!(action, VsrAction::CommitJournal))
+ };
+ if started {
+ let config = partitions.config();
+ if let Some(partition) = partitions.get_mut_by_ns(&namespace) {
+ partition.commit_journal(config).await;
+ }
}
}
- #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
- async fn send_request_state_transfer<P>(
- &self,
- consensus: &VsrConsensus<B, P>,
- target: u8,
- nonce: u128,
- ) where
+ /// Re-request the remaining repair window when the stream has gone quiet.
+ ///
+ /// Repair frames are fire-and-forget, so a lost one leaves the session
armed
+ /// forever with the commit walk pinned below the frontier.
+ #[allow(clippy::future_not_send)]
+ async fn retry_stalled_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>)
+ where
B: MessageBus,
P: Pipeline<Entry = consensus::PipelineEntry>,
{
- let msg =
-
Message::<RequestStateTransferHeader>::new(size_of::<RequestStateTransferHeader>())
- .transmute_header(|_, h: &mut RequestStateTransferHeader| {
- h.command = Command2::RequestStateTransfer;
- h.cluster = consensus.cluster();
- h.replica = consensus.replica();
- h.nonce = nonce;
- h.namespace = consensus.namespace();
- h.size = size_of::<RequestStateTransferHeader>() as u32;
- });
- let _ = self
- .bus
- .send_to_replica(target, msg.into_generic().into_frozen())
- .await;
+ // Stall retry (mirrors `tick_partitions`): a lost frame must not
wedge it.
+ let repair_retry_ticks = self.repair_retry_ticks.get();
+ let stalled = {
+ // `ViewChange` too: a parked view change repairs toward its
merged log
+ // and cannot start until the window fills. Gating on `Normal`
alone
+ // defers a dropped frame to the 500-tick escalation.
+ let repairing_view = consensus.pending_view_log().is_some()
+ && consensus.is_primary_for_view(consensus.view());
+ let mut session = self.metadata_repair.borrow_mut();
+ session.as_mut().and_then(|session| {
+ if !consensus.is_normal() && !repairing_view {
+ return None;
+ }
+ session.idle_ticks += 1;
+ if session.idle_ticks < repair_retry_ticks {
+ return None;
+ }
+ session.idle_ticks = 0;
+ Some((session.peer, session.nonce, session.to_op))
+ })
+ };
+ if let Some((peer, nonce, to_op)) = stalled {
+ // Primary-elect only. Its window starts at the merged log's commit
+ // point, which can sit below local `commit_min` (the headers
inherited
+ // from senders behind the canonical log_view live there), so
+ // `commit_min + 1` would skip them. A backup's parked `StartView`
+ // suffix is only a verification reference; resuming from its
commit
+ // point would restart at the view's opening head, not at the gap.
+ let from_op = consensus
+ .pending_view_log()
+ .filter(|_| consensus.is_primary_for_view(consensus.view()))
+ .map_or_else(
+ || consensus.commit_min() + 1,
+ |pending| pending.commit_max.max(1),
+ );
+ if from_op <= to_op {
+ tracing::info!(
+ shard = self.id,
+ from_op,
+ to_op,
+ peer,
+ "metadata repair stalled; re-requesting remaining window"
+ );
+ self.send_request_prepares(
+ consensus.cluster(),
+ consensus.replica(),
+ peer,
+ nonce,
+ from_op,
+ to_op,
+ consensus.namespace(),
+ )
+ .await;
+ }
+ }
}
- /// Answer a `RequestStateTransfer`: `offer = None` sends a header-only
- /// `available = 0` (the requester falls back to journal repair or
- /// retries elsewhere); an offer ships its encoded state manifest as the
- /// frame body.
- #[allow(
- clippy::future_not_send,
- clippy::cast_possible_truncation,
- clippy::too_many_arguments
- )]
- async fn send_state_transfer_target(
- &self,
- cluster: u128,
- self_id: u8,
- target: u8,
- nonce: u128,
- namespace: u64,
- descriptor: TransferDescriptor<'_>,
- ) where
+ /// Compare a backup's log against the headers the concluding `StartView`
+ /// published, and report where they disagree.
+ ///
+ /// Without this, divergence is silent and permanent: the backup acks with
its
+ /// own checksum, the primary rejects the ack, and journal repair skips an
op
+ /// it already has a header for.
+ ///
+ /// The split at the announced commit point is what matters. Above it a
+ /// disagreement is ordinary, so the entry is dropped and the primary's
+ /// retransmission refills the range. At or below it, this replica applied
+ /// something the view says was different, which only state transfer
fixes, so
+ /// it is reported and left alone.
+ ///
+ /// Truncation uses `Journal::truncate_from`, not `drain`: `drain` advances
+ /// `snapshot_op` past what it removed, marking ops that must stay
refillable
+ /// as evictable.
+ #[allow(clippy::future_not_send)]
+ async fn reconcile_metadata_view_divergence(&self)
+ where
B: MessageBus,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target: Journal<
+ <MJ as JournalHandle>::Storage,
+ Entry = Message<PrepareHeader>,
+ Header = PrepareHeader,
+ >,
+ M: MetadataStm,
{
- let manifest = descriptor
- .offer
- .map(|(entries, _)| consensus::encode_state_manifest(entries));
- let total_size =
- size_of::<StateTransferTargetHeader>() +
manifest.as_ref().map_or(0, Vec::len);
- let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
- if let Some(manifest) = &manifest {
-
msg.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(manifest);
+ let metadata = self.plane.metadata();
+ let Some(ref consensus) = metadata.consensus else {
+ return;
+ };
+ // Backups only; a primary reconciles through the merge itself.
+ if consensus.is_primary_for_view(consensus.view()) {
+ return;
}
- let msg = msg.transmute_header(|_, h: &mut StateTransferTargetHeader| {
- h.command = Command2::StateTransferTarget;
- h.cluster = cluster;
- h.replica = self_id;
- h.nonce = nonce;
- h.namespace = namespace;
- h.size = total_size as u32;
- // The serving replica's own progress travels with every
descriptor,
- // available or not: it is what lets a receiver refuse an offer
from
- // a replica that knows less than it does.
- h.view = descriptor.view;
- h.commit_max = descriptor.commit_max;
- h.unavailable_transient = u8::from(descriptor.transient);
- if let Some((_, commit_op)) = descriptor.offer {
- h.available = 1;
- h.commit_op = commit_op;
- }
- });
- let _ = self
- .bus
- .send_to_replica(target, msg.into_generic().into_frozen())
- .await;
- }
+ let Some(pending) = consensus.pending_view_log() else {
+ return;
+ };
+ let Some(journal) = metadata.journal.as_ref() else {
+ return;
+ };
- #[allow(
- clippy::future_not_send,
- clippy::cast_possible_truncation,
+ // Truncation is safe only above what this replica has *applied*,
which is
+ // not the view's commit point: `pending.commit_max` is the new
primary's
+ // number and a backup can sit above it. Splitting on the view's number
+ // would drop already-executed ops with no rollback, and silently.
+ let applied_floor = pending.commit_max.max(consensus.commit_min());
+
+ let mut repairable_from: Option<u64> = None;
Review Comment:
the reconcile only walks ops named in `pending.headers`, which spans
commit..head - a local WAL suffix *above* the announced head is never
truncated, while `set_sequence(announced)` drops the head under it. the next
prepare at `op_head+1` then hits the append slot-collision error (it fires even
when `existing.op == header.op`) and the backup wedges. that's the case the
TODO in `handle_start_view` describes, and `truncate_from` is the tool for it -
just not wired to this case. the subtle part: `repairable_from` only arms on an
in-window checksum disagreement, so the miss is precisely the log that *agrees*
in-window and is only stale above the head - the restart/re-adopt shape. needs
a `journal last_op > pending.op_head` check, which wants a trait accessor
(`last_op` is inherent on `PrepareJournal`).
##########
core/binary_protocol/src/consensus/header.rs:
##########
@@ -78,12 +105,82 @@ pub trait ConsensusHeader: Sized + CheckedBitPattern +
NoUninit {
command == Self::COMMAND
}
+ /// Whether this header's `checksum` field seals the frame.
+ ///
+ /// True for replica-to-replica control frames, whose header carries every
+ /// decision field: view number, commit point, and the nack bitset that
+ /// authorises truncation. TCP's 16-bit checksum does not reliably catch a
+ /// flipped bit on a plaintext replica link.
+ ///
+ /// False for three groups: [`PrepareHeader`] / [`RepairPrepareHeader`]
spend
+ /// `checksum` on [`PrepareHeader::identity_checksum`], which excludes
`view` so
+ /// a re-stamped prepare keeps one identity, and a seal cannot share the
field;
+ /// [`RequestHeader`] / [`ReplyHeader`] / [`EvictionHeader`] cross the
client
+ /// boundary, so sealing them is an SDK change on both ends;
[`GenericHeader`] is
+ /// the type-erased pre-dispatch view and defers to the typed parse, where
+ /// [`Self::verify_frame`] runs.
+ const FRAME_SEALED: bool = true;
+
/// # Errors
/// Returns `ConsensusError` if the header fields are inconsistent.
fn validate(&self) -> Result<(), ConsensusError>;
fn operation(&self) -> Operation;
fn command(&self) -> Command2;
fn size(&self) -> u32;
+
+ /// The `checksum` field, whatever this header spends it on.
+ fn checksum(&self) -> u128;
+
+ /// Overwrite the `checksum` field.
+ fn set_checksum(&mut self, checksum: u128);
+
+ /// Checksum over every byte of the header past `checksum` itself.
+ ///
+ /// `checksum_body` sits inside that range, so sealing the header also
+ /// pins the body seal, and the two together cover the whole frame.
+ #[must_use]
+ fn frame_checksum(&self) -> u128 {
+ let bytes: &[u8; HEADER_SIZE] = bytemuck::bytes_of(self)
+ .try_into()
+ .expect("every consensus header is HEADER_SIZE bytes");
+ frame_checksum_bytes(bytes)
+ }
+
+ /// Stamp [`Self::frame_checksum`]. Call last when building a frame: it
covers
+ /// every other field, `checksum_body` included, so later writes are
uncovered.
+ fn seal(&mut self) {
Review Comment:
calling `seal()` on a `FRAME_SEALED = false` type (a prepare) would silently
overwrite the identity checksum with a frame checksum in release - the
`debug_assert` is the only guard. no prepare goes through `seal()` today
(`seal_prepare_checksum` writes the field directly), but making `FRAME_SEALED`
a required const with no trait default turns that mistake into a compile error
for free.
##########
core/consensus/src/impls.rs:
##########
@@ -2818,6 +3137,110 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
// the loopback queue directly.
self.loopback_queue.borrow_mut().clear();
+ tracing::info!(
+ replica = self.replica,
+ view = self.view.get(),
+ op_head = new_op,
+ commit_max = max_commit,
+ "view-change quorum merged; repairing up to the merged log before
starting the view"
+ );
+ emit_replica_event(
+ SimEventKind::ReplicaStateChanged,
+ &ReplicaLogContext::from_consensus(self, plane),
+ );
+
+ // No sends yet: `SendStartView` promises this replica can serve every
op in
+ // the merged log, and a backup adopting the announced head asks it
for the
+ // bodies behind it.
+ Vec::new()
+ }
+
+ /// Sizes handed to the DVC merge.
+ const fn merge_quorums(&self) -> MergeQuorums {
+ MergeQuorums {
+ view_change: self.quorum_view_change(),
+ nack_prepare: self.quorum_nack_prepare(),
+ replica_count: self.replica_count as usize,
+ // The cluster-wide ceiling, not `self.prepare_queue_max`: this
node's
+ // config says nothing about how deep a peer's pipeline is.
+ prepare_queue_ceiling: PREPARE_QUEUE_CEILING as u64,
+ }
+ }
+
+ /// The merged log this replica is repairing toward, if a view change is
+ /// mid-transition. The shard reads it for the op range it must cover
before the
+ /// view can start, and for which peers offered the bodies.
+ #[must_use]
+ pub fn pending_view_log(&self) -> Option<MergedLog> {
Review Comment:
`pending_view_log()` clones the whole `MergedLog` (two `Vec<PrepareHeader>`)
on every call, and most call sites want `.is_some()` (every metadata tick) or
two u64s; the repair path clones once per repaired prepare. an `is_some()`
accessor + a bounds accessor + a with-ref closure covers all but the two sites
that hold the value across an await or across `start_pending_view` (which takes
the cell and would trip the RefCell inside a closure).
##########
core/consensus/src/view_change_quorum.rs:
##########
@@ -64,43 +239,196 @@ pub fn dvc_count(array: &DvcQuorumArray) -> usize {
array.iter().filter(|m| m.is_some()).count()
}
-/// Check if a specific replica has sent a DVC.
-#[must_use]
-pub fn dvc_has_from(array: &DvcQuorumArray, replica: u8) -> bool {
- array.get(replica as usize).is_some_and(Option::is_some)
+/// Reset the DVC quorum array.
+pub fn dvc_reset(array: &mut DvcQuorumArray) {
+ *array = dvc_quorum_array_empty();
}
-/// Select the winning DVC (best log) from the quorum.
-/// Returns the DVC with: highest `log_view`, then highest op.
-#[must_use]
-pub fn dvc_select_winner(array: &DvcQuorumArray) -> Option<&StoredDvc> {
- array
- .iter()
- .filter_map(|m| m.as_ref())
- .max_by(|a, b| match a.log_view.cmp(&b.log_view) {
- std::cmp::Ordering::Equal => a.op.cmp(&b.op),
- other => other,
- })
+/// Iterator over all stored DVCs.
+pub fn dvc_iter(array: &DvcQuorumArray) -> impl Iterator<Item = &StoredDvc> {
+ array.iter().filter_map(|m| m.as_ref())
}
-/// Get the maximum commit number across all DVCs.
-#[must_use]
-pub fn dvc_max_commit(array: &DvcQuorumArray) -> u64 {
- array
- .iter()
- .filter_map(|m| m.as_ref())
- .map(|dvc| dvc.commit)
- .max()
- .unwrap_or(0)
+/// Why a `DoViewChange` body could not be read as a suffix.
+///
+/// Dropped whole rather than partially trusted: the merge indexes
arithmetically
+/// from the head op, so one bad offset misattributes a header, nack, or body.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum DvcSuffixError {
+ /// Body length is not a whole number of headers.
+ NotHeaderMultiple { body_len: usize },
+ /// More entries than the bitsets can address.
+ TooManyEntries { count: usize },
+ /// An entry is not a valid `PrepareHeader` bit pattern.
+ MalformedHeader { index: usize },
+ /// Entries are not consecutive descending from the head op.
+ OpOutOfOrder {
+ index: usize,
+ expected: u64,
+ found: u64,
+ },
+ /// A bitset addresses an entry the body does not contain.
+ BitsetBeyondSuffix { count: usize },
+ /// An entry's identity checksum does not match its own contents.
+ ChecksumMismatch { index: usize },
+ /// A lower entry claims a view newer than the entry above it.
+ ViewRegressed { index: usize },
+ /// A lower entry claims a timestamp at or after the entry above it.
+ TimestampNotDecreasing { index: usize },
+ /// Consecutive entries do not hash-chain.
+ ChainBreak { index: usize },
}
-/// Reset the DVC quorum array.
-pub const fn dvc_reset(array: &mut DvcQuorumArray) {
- *array = dvc_quorum_array_empty();
+impl std::fmt::Display for DvcSuffixError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::NotHeaderMultiple { body_len } => write!(
+ f,
+ "do_view_change body of {body_len} bytes is not a whole number
of {} -byte headers",
+ size_of::<PrepareHeader>()
+ ),
+ Self::TooManyEntries { count } => write!(
+ f,
+ "do_view_change suffix of {count} entries exceeds the maximum
{DVC_HEADERS_MAX}"
+ ),
+ Self::MalformedHeader { index } => {
+ write!(
+ f,
+ "do_view_change suffix entry {index} is not a prepare
header"
+ )
+ }
+ Self::OpOutOfOrder {
+ index,
+ expected,
+ found,
+ } => write!(
+ f,
+ "do_view_change suffix entry {index} carries op {found},
expected {expected}"
+ ),
+ Self::BitsetBeyondSuffix { count } => write!(
+ f,
+ "do_view_change bitset addresses an entry past the
{count}-entry suffix"
+ ),
+ Self::ChecksumMismatch { index } => write!(
+ f,
+ "do_view_change suffix entry {index} does not match its own
identity checksum"
+ ),
+ Self::ViewRegressed { index } => write!(
+ f,
+ "do_view_change suffix entry {index} claims a newer view than
the entry above it"
+ ),
+ Self::TimestampNotDecreasing { index } => write!(
+ f,
+ "do_view_change suffix entry {index} does not predate the
entry above it"
+ ),
+ Self::ChainBreak { index } => write!(
+ f,
+ "do_view_change suffix entry {index} does not chain to the
entry above it"
+ ),
+ }
+ }
}
-/// Iterator over all stored DVCs.
-// TODO: add #[must_use] -- pure iterator query, callers should not ignore.
-pub fn dvc_iter(array: &DvcQuorumArray) -> impl Iterator<Item = &StoredDvc> {
- array.iter().filter_map(|m| m.as_ref())
+impl std::error::Error for DvcSuffixError {}
+
+/// Read a suffix out of a `DoViewChange` body.
+///
+/// `head_op` is the sender's `header.op`; entries run consecutively down from
it,
+/// blanks included, so slot `i` is unambiguously op `head_op - i`. An empty
body
+/// yields an empty suffix, which is what a sender with nothing to describe
sends.
+///
+/// `body` must carry [`PrepareHeader`]'s alignment: the cast below is
checked, so an
+/// unaligned body reports every entry as [`DvcSuffixError::MalformedHeader`]
instead
+/// of what is actually wrong. Real frames clear this because the body starts
a whole
+/// number of 256-byte headers into an aligned buffer; the debug assert
catches a
+/// hand-built one.
+///
+/// # Errors
+/// [`DvcSuffixError`] when the body is not a consecutive run of valid prepare
+/// headers descending from `head_op`, or a bitset addresses a missing entry.
+pub fn dvc_suffix_decode(
+ body: &[u8],
+ head_op: u64,
+ nack_bitset: u128,
+ present_bitset: u128,
+) -> Result<DvcSuffix, DvcSuffixError> {
+ debug_assert!(
+ body.is_empty()
+ || body
+ .as_ptr()
+ .addr()
+ .is_multiple_of(align_of::<PrepareHeader>()),
+ "suffix body must be aligned for PrepareHeader"
+ );
+ let header_size = size_of::<PrepareHeader>();
+ if !body.len().is_multiple_of(header_size) {
+ return Err(DvcSuffixError::NotHeaderMultiple {
+ body_len: body.len(),
+ });
+ }
+ let count = body.len() / header_size;
+ if count > DVC_HEADERS_MAX {
+ return Err(DvcSuffixError::TooManyEntries { count });
+ }
+ if count < DVC_HEADERS_MAX {
+ let beyond = !((1u128 << count) - 1);
+ if nack_bitset & beyond != 0 || present_bitset & beyond != 0 {
+ return Err(DvcSuffixError::BitsetBeyondSuffix { count });
+ }
+ }
+
+ let mut headers = Vec::with_capacity(count);
+ // The entry above the current one, skipping blanks: high-to-low, so the
child.
+ let mut child: Option<PrepareHeader> = None;
+ for index in 0..count {
+ let chunk = &body[index * header_size..(index + 1) * header_size];
+ let header = bytemuck::checked::try_from_bytes::<PrepareHeader>(chunk)
+ .map_err(|_| DvcSuffixError::MalformedHeader { index })?;
+ let expected = head_op
+ .checked_sub(index as u64)
+ .ok_or(DvcSuffixError::OpOutOfOrder {
+ index,
+ expected: 0,
+ found: header.op,
+ })?;
+ if header.op != expected {
+ return Err(DvcSuffixError::OpOutOfOrder {
+ index,
+ expected,
+ found: header.op,
+ });
+ }
+
+ if matches!(dvc_header_kind(header), DvcHeaderKind::Valid) {
+ // Recompute rather than trust the field. Otherwise one bit
flipped in
+ // transit becomes a canonical header no replica holds, honest
senders
+ // read as disagreeing, and a corrupted frame turns into a nack
quorum
+ // against a committed op. A pre-sealing peer's sentinel is
skipped.
+ if header.checksum != CHECKSUM_UNSEALED &&
header.identity_checksum() != header.checksum
+ {
+ return Err(DvcSuffixError::ChecksumMismatch { index });
+ }
+ if let Some(child) = child {
+ // Views never go backwards down the log, timestamps never
forwards,
+ // and consecutive entries hash-chain. A frame breaking any of
these
+ // describes a log that cannot exist.
+ if header.view > child.view {
+ return Err(DvcSuffixError::ViewRegressed { index });
Review Comment:
the `ViewRegressed` check rejects a log this codebase legitimately produces.
`restamp_prepare_view` rewrites `view` on the wire copy only, and
`on_replicate` skips the append when the journal already holds the op - so a
backup that held op 12 from view 1 but missed 11 and 13 appends those two at
view 2 and keeps 12 at view 1. its next suffix decodes [13:v2, 12:v1, 11:v2],
trips this check, and the whole `DoViewChange` is dropped before the view
advance; every retransmit is byte-identical, so it drops every time. if that
replica is needed for the view-change quorum, the view change stalls until
something else moves.
gating on consecutiveness like `ChainBreak` below doesn't help - 11 and 12
are adjacent in that scenario. the premise "views never go backwards down the
log" doesn't hold once skip-kept entries exist; this check has to go or be
rethought. `TimestampNotDecreasing` next to it can fire across a skip-kept
entry the same way - worth checking with the same scenario.
##########
core/consensus/src/dvc_merge.rs:
##########
@@ -0,0 +1,911 @@
+// 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.
+
+//! Merging a `DoViewChange` quorum into the new view's log: for every op that
+//! might be uncommitted, does the new view keep it or discard it?
+//!
+//! Keeping an op that was never committed costs a wasted slot. Discarding one
+//! that WAS committed loses acknowledged client data, so the only proof
accepted
+//! for discarding is a nack quorum: enough replicas stating they never
prepared
+//! it that a replication quorum provably never formed. Absent that the op is
+//! kept, and if no replica offers its body the view does not start. Stalling
is
+//! visible and recoverable; losing the op is neither.
+
+#[cfg(test)]
+use crate::view_change_quorum::DvcSuffix;
+use crate::view_change_quorum::{DvcQuorumArray, StoredDvc, dvc_count,
dvc_iter};
+use iggy_binary_protocol::PrepareHeader;
+
+/// Sizes the merge needs from the replica.
+#[derive(Debug, Clone, Copy)]
+pub struct MergeQuorums {
+ /// `DoViewChange` messages needed before a view may start.
+ pub view_change: usize,
+ /// Nacks needed to prove an op uncommitted, so it may be discarded.
+ pub nack_prepare: usize,
+ /// Cluster size, which bounds how many more DVCs could still arrive.
+ pub replica_count: usize,
+ /// Cluster-wide pipeline ceiling: an op further than this below a
sender's head
+ /// cannot still be uncommitted, since no node could have kept it in
flight.
+ ///
+ /// NOT this node's configured depth. The bound applies to a *peer's* head
op,
+ /// and a local depth larger than that peer's manufactures a commit the
peer
+ /// never made. Every node's depth is pinned below `DVC_HEADERS_MAX`, so
the
+ /// ceiling holds for all of them.
+ pub prepare_queue_ceiling: u64,
+}
+
+/// What the collected DVCs say about starting the view.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum MergeOutcome {
+ /// Fewer than `view_change` DVCs so far.
+ AwaitingQuorum,
+ /// Quorum is in, but some op is neither provably dead nor recoverable and
an
+ /// unreported replica could still settle it. Wait for them.
+ AwaitingRepair {
+ /// The op that cannot yet be decided.
+ undecided_op: u64,
+ },
+ /// Every replica reported and an op is still neither provably dead nor
+ /// recoverable. No further message changes that: data loss already
happened,
+ /// and truncating here would turn it from detected into silent.
+ Deadlocked {
+ /// The op that cannot be decided.
+ undecided_op: u64,
+ },
+ /// The view can start.
+ Ready(MergedLog),
+}
+
+/// The log the new primary adopts.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MergedLog {
+ /// New head op. Below the highest op any canonical sender reported when a
+ /// nack quorum proved the ops above it dead.
+ pub op_head: u64,
+ /// Highest op the quorum proves committed. The merge never discards at or
+ /// below this.
+ pub commit_max: u64,
+ /// Canonical headers for `commit_max..=op_head`, ordered high-to-low op.
+ /// The new primary installs these over its own log.
+ pub headers: Vec<PrepareHeader>,
+ /// Headers non-canonical senders report committed. Installed
unconditionally,
+ /// because header repair will not cross a gap to reach them later.
+ pub committed_elsewhere: Vec<PrepareHeader>,
+}
+
+/// Highest op the quorum proves committed.
+///
+/// Three independent lower bounds, because a single sender's view of the
commit
+/// point can lag arbitrarily while the cluster's cannot:
+/// * each sender's own reported commit,
+/// * the `commit` its head prepare carries, stamped by that op's primary,
+/// * its head minus the cluster-wide pipeline ceiling, since nothing further
back
+/// than one pipeline can still be in flight.
+///
+/// The lowest op in a sender's suffix is deliberately NOT a fourth bound, and
+/// re-adding one is a data-loss bug. It would be sound only if the suffix
floor
+/// were the commit point by construction; here it is computed, and two paths
in
+/// `build_dvc_suffix` raise it above the sender's commit (ops are 1-based, so
+/// commit 0 floors at op 1; the `DVC_HEADERS_MAX` clamp drops the bottom of an
+/// over-wide window). Nothing on the wire distinguishes the two.
+///
+/// Nothing is lost by omitting it: the snapshot is tagged with the `(op,
commit)`
+/// the header is stamped with and dropped on a mismatch, so the floor either
+/// equals `dvc.commit`, already the first bound, or exceeds it, the unsound
case.
+#[must_use]
+pub fn merge_commit_max(quorum: &DvcQuorumArray, prepare_queue_ceiling: u64)
-> u64 {
+ let mut commit_max = 0;
+ for dvc in dvc_iter(quorum) {
+ commit_max = commit_max.max(dvc.commit);
+ commit_max =
commit_max.max(dvc.op.saturating_sub(prepare_queue_ceiling));
+ if let Some(head) = dvc.suffix.headers().first() {
+ commit_max = commit_max.max(head.commit);
+ }
+ }
+ commit_max
+}
+
+/// Highest `log_view` any sender reported.
+///
+/// Senders at this `log_view` were in every earlier view change, so their
headers
+/// already reflect the truncations those views decided. That makes them
canonical,
+/// and a lower-`log_view` sender disagreeing is evidence against its own
header.
+fn log_view_canonical(quorum: &DvcQuorumArray) -> Option<u32> {
+ dvc_iter(quorum).map(|dvc| dvc.log_view).max()
+}
+
+/// Per-op tally over the whole quorum.
+struct OpVerdict<'a> {
+ canonical: Option<&'a PrepareHeader>,
+ /// Senders holding the canonical header AND able to serve its body.
+ copies: usize,
+ nacks: usize,
+ /// Canonical senders disagree about this op, so no header here is
trustworthy.
+ conflict: bool,
+}
+
+/// What the canonical senders say about one op.
+struct CanonicalAt<'a> {
+ header: Option<&'a PrepareHeader>,
+ /// Two senders at the canonical `log_view` disagree here.
+ conflict: bool,
+}
+
+/// The canonical header at `op`, and whether the canonical senders agree.
+///
+/// Every canonical sender is consulted, not just the first: they were all in
+/// normal status in that view and a primary prepares one thing per op, so a
+/// disagreement is not a vote but proof that one header is wrong with no way
to
+/// tell which. The caller treats the op as undecidable.
+fn canonical_header_at<'a>(canonical_senders: &[&'a StoredDvc], op: u64) ->
CanonicalAt<'a> {
+ let mut header: Option<&'a PrepareHeader> = None;
+ let mut conflict = false;
+ for dvc in canonical_senders {
+ let Some(index) = dvc.suffix.index_of(dvc.op, op) else {
+ continue;
+ };
+ let Some(candidate) = dvc.suffix.valid_header_at(index) else {
+ continue;
+ };
+ match header {
+ Some(existing) if existing.checksum != candidate.checksum =>
conflict = true,
+ Some(_) => {}
+ None => header = Some(candidate),
+ }
+ }
+ CanonicalAt { header, conflict }
+}
+
+/// Tally every sender's position on `op`.
+fn tally_op<'a>(
+ quorum: &'a DvcQuorumArray,
+ canonical_senders: &[&'a StoredDvc],
+ canonical_log_view: u32,
+ op: u64,
+) -> OpVerdict<'a> {
+ let CanonicalAt {
+ header: canonical,
+ conflict,
+ } = canonical_header_at(canonical_senders, op);
+ let mut copies = 0;
+ let mut nacks = 0;
+
+ for dvc in dvc_iter(quorum) {
+ // The sender's log stops below this op, so it never prepared it.
+ if dvc.op < op {
+ nacks += 1;
+ continue;
+ }
+ let Some(index) = dvc.suffix.index_of(dvc.op, op) else {
+ // The sender said nothing about this op, so it abstains: no nack,
no
+ // copy. Counting silence as a nack is sound only when every DVC
carries
+ // headers, since then falling outside a window means being above
it.
+ // Two cases here are not, and abstaining costs a slower view
change
+ // where nacking costs data.
+ //
+ // An empty suffix is not a vote. It comes from a replica with
+ // nothing uncommitted, or one whose snapshot no longer matches its
+ // log; reading it as
+ // agreement with someone else's nack discards a committed op on
one real
+ // nack plus one silence. And with `dvc.op >= op` established
above, a
+ // non-empty suffix not covering `op` puts `op` below the sender's
window
+ // floor, at or below its own commit point, so nacking it is
backwards.
+ // Only the defensive `DVC_HEADERS_MAX` clamp reaches that.
+ continue;
+ };
+
+ let held = dvc.suffix.valid_header_at(index);
+ if let (Some(held), Some(canonical)) = (held, canonical)
+ && dvc.suffix.offers_body(index)
+ && held.checksum == canonical.checksum
+ {
+ copies += 1;
+ }
+
+ if dvc.suffix.nacks(index) {
+ // Explicit: the sender proves it never prepared this op.
+ nacks += 1;
+ } else if let Some(held) = held {
+ // Only a sender BEHIND the canonical log_view can implicitly nack.
+ // Without this, corrupting one canonical header in transit turns
every
+ // honest sender's correct header into an implicit nack against the
+ // garbage: a nack quorum on three replicas. A same-log_view
+ // disagreement is evidence, not a vote, and goes through
`conflict`.
+ let may_nack_implicitly = dvc.log_view < canonical_log_view;
+ match canonical {
+ // Implicit: the sender holds a DIFFERENT prepare, so not this
one.
+ Some(canonical) if may_nack_implicitly && held.checksum !=
canonical.checksum => {
+ nacks += 1;
+ }
+ // Implicit: no canonical sender holds anything here, so a
newer
+ // view already truncated this op and the sender holds a
corpse.
+ None if may_nack_implicitly => nacks += 1,
+ _ => {}
+ }
+ }
+ }
+
+ OpVerdict {
+ canonical,
+ copies,
+ nacks,
+ conflict,
+ }
+}
+
+/// Collect the canonical headers for `commit_max..=op_head`, high-to-low.
+///
+/// Checks the hash chain as it walks: a break means the canonical senders
agreed
+/// on individual ops but not on one history, which no later step would notice.
+fn canonical_headers(
+ canonical_senders: &[&StoredDvc],
+ op_head: u64,
+ commit_max: u64,
+) -> Option<Vec<PrepareHeader>> {
+ if op_head == 0 {
+ return Some(Vec::new());
+ }
+ let floor = commit_max.max(1);
+ let mut headers = Vec::new();
+ let mut child: Option<PrepareHeader> = None;
+ let mut op = op_head;
+ loop {
+ let at = canonical_header_at(canonical_senders, op);
+ if at.conflict {
+ return None;
+ }
+ let header = *at.header?;
+ if let Some(child) = child
+ && child.parent != header.checksum
+ {
+ tracing::error!(
+ op,
+ child_op = child.op,
+ "view-change headers do not hash-chain; refusing to install"
+ );
+ return None;
+ }
+ child = Some(header);
+ headers.push(header);
+ if op <= floor {
+ break;
+ }
+ op -= 1;
+ }
+ Some(headers)
+}
+
+/// Headers that a non-canonical sender reports committed.
+///
+/// Trusted on that sender's word alone, unlike anything above its commit
point:
+/// header repair walks the chain backwards and stops at a gap, so an op
missing
+/// below the new primary's commit point can never be repaired into place and
+/// refusing it here strands the log permanently. The claim is already
+/// quorum-backed, since a sender cannot report an op committed unless a
+/// replication quorum held it.
+///
+/// `None` when two senders report *different* prepares committed at one op.
These
+/// install unconditionally, so guessing is least affordable here; as in
+/// [`canonical_header_at`], the caller treats it as undecidable.
+fn committed_elsewhere(
Review Comment:
this takes one stale sender's word for it - no second sender, no nack
quorum, no chain check against the canonical range. `dvc_commit` reports
`commit_max.min(op)`, `commit_max` advances from the primary's claim without
checking the local log matches, and reconcile deliberately leaves a divergent
entry at or below `applied_floor` in place - so a replica can report commit >=
X while holding a header that never committed, and ship it here on the next
view change. nothing installs these headers into the log, but they poison the
repair gate: the genuine repaired prepare then mismatches `expected.checksum`
and gets discarded, and the view change stalls at that op. wants corroboration
or a chain check before trusting the entry.
##########
core/shard/src/lib.rs:
##########
@@ -7438,10 +7914,411 @@ 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.
+///
+/// Read through `repair_header`, not the resident headers: the committed
prefix
+/// leaves those as soon as its bytes reach a segment, which on a caught-up
+/// replica includes the commit point itself.
+fn refresh_partition_dvc_suffix<B, SB>(partition:
&partitions::IggyPartition<B, SB>)
+where
+ B: MessageBus,
+ SB: SuperblockStore,
+{
+ 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.repair_header(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() {
Review Comment:
the partition arm's `header_at` is `repair_header` - a linear `header_by_op`
scan plus a linear walk of the evicted ring - probed per op across the window,
so up to 128 x resident-header-count comparisons per rebuild.
`repaired_window_shape`'s doc warns about exactly this shape, and the tick-gate
comment nearby bounds the *frequency*, not the per-rebuild cost: this reruns on
every SVC/DVC arrival and every non-Normal tick per partition
(`local_dvc_suffix_stale` keys on `(op, commit)`, and repair advances it), so
an active view change burns it repeatedly on the pump. a one-pass window
builder over headers + ring, same shape as `repaired_window_shape`, kills the
quadratic.
##########
core/binary_protocol/src/consensus/header.rs:
##########
@@ -1023,10 +1297,52 @@ impl ConsensusHeader for DoViewChangeHeader {
"commit cannot exceed op".to_string(),
));
}
+ let suffix_len = self.suffix_len()?;
+ // Bits past the suffix describe entries never sent: unchecked, a peer
could
+ // smuggle a nack for an op the new primary would then truncate.
+ if suffix_len < DVC_HEADERS_MAX {
+ let beyond = !((1u128 << suffix_len) - 1);
+ if self.nack_bitset & beyond != 0 || self.present_bitset & beyond
!= 0 {
+ return Err(ConsensusError::InvalidField(format!(
+ "do_view_change: bitset bits set past the
{suffix_len}-entry suffix"
+ )));
+ }
+ }
Ok(())
}
}
+impl DoViewChangeHeader {
+ /// Number of `PrepareHeader`s in the body.
+ ///
+ /// Zero is valid and means "no suffix": a replica with nothing
uncommitted, or a
+ /// peer predating the suffix. Both contribute numbers only.
+ ///
+ /// # Errors
+ /// [`ConsensusError::InvalidField`] when `size` is short of the header,
is not a
+ /// whole number of headers, or exceeds what the bitsets can address.
+ pub fn suffix_len(&self) -> Result<usize, ConsensusError> {
Review Comment:
body is token-identical to `StartViewHeader::suffix_len` except the label in
the three messages - one free fn taking the frame name, two thin wrappers, and
the bounds logic can't drift apart.
##########
core/simulator/src/deps.rs:
##########
@@ -180,6 +180,33 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal<S> for
SimJournal<S> {
/// ever superseded by a snapshot. Answered explicitly (the trait has no
/// default) so a simulated state transfer has to opt into a watermark
/// rather than silently inherit one that never moves.
+ /// Drop the suffix, so a simulated backup whose entries disagree with a
started
+ /// view reconciles the way a real one does. Mirrors
+ /// `PrepareJournal::truncate_from`, whose watermark stays put; here it
never moves.
+ async fn truncate_from(&self, from_op: u64) -> std::io::Result<usize> {
Review Comment:
this landed between `snapshot_op`'s doc comment and `snapshot_op` itself -
the doc above now describes the wrong method and `snapshot_op` lost its doc.
move it below `set_snapshot_op`.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -98,17 +178,43 @@ async fn read_persisted_offset(path: &str) ->
Result<Option<u64>, IggyError> {
.open(path)
.await
.map_err(|_|
IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
- let buf = vec![0u8; OFFSET_SIZE];
- let compio::BufResult(read, buf) = file.read_exact_at(buf, 0).await;
- match read {
- Ok(()) => {}
- Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof =>
return Ok(None),
+ // Read the whole record, falling back to a bare offset: a pre-checksum
file is
+ // exactly `OFFSET_SIZE` long, so the first read reports EOF rather than
failing.
+ let compio::BufResult(read, buf) = file.read_exact_at(vec![0u8;
OFFSET_RECORD_SIZE], 0).await;
+ let bytes = match read {
+ Ok(()) => buf,
+ Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
Review Comment:
the 16-then-8 fallback costs a second syscall + alloc per legacy-file read
just to learn the length. `decode_offset_record` already classifies any length,
so one non-exact `read_at` into a 16-byte buffer and decoding `&buf[..n]` does
it. the `..n` matters: decoding the zero-padded buffer would turn a legacy
8-byte file into `Corrupt`, and that's a hard `Err` out of
`read_persisted_offset`, not a `None`.
##########
core/consensus/src/view_change_quorum.rs:
##########
@@ -64,43 +239,196 @@ pub fn dvc_count(array: &DvcQuorumArray) -> usize {
array.iter().filter(|m| m.is_some()).count()
}
-/// Check if a specific replica has sent a DVC.
-#[must_use]
-pub fn dvc_has_from(array: &DvcQuorumArray, replica: u8) -> bool {
- array.get(replica as usize).is_some_and(Option::is_some)
+/// Reset the DVC quorum array.
+pub fn dvc_reset(array: &mut DvcQuorumArray) {
+ *array = dvc_quorum_array_empty();
}
-/// Select the winning DVC (best log) from the quorum.
-/// Returns the DVC with: highest `log_view`, then highest op.
-#[must_use]
-pub fn dvc_select_winner(array: &DvcQuorumArray) -> Option<&StoredDvc> {
- array
- .iter()
- .filter_map(|m| m.as_ref())
- .max_by(|a, b| match a.log_view.cmp(&b.log_view) {
- std::cmp::Ordering::Equal => a.op.cmp(&b.op),
- other => other,
- })
+/// Iterator over all stored DVCs.
+pub fn dvc_iter(array: &DvcQuorumArray) -> impl Iterator<Item = &StoredDvc> {
+ array.iter().filter_map(|m| m.as_ref())
}
-/// Get the maximum commit number across all DVCs.
-#[must_use]
-pub fn dvc_max_commit(array: &DvcQuorumArray) -> u64 {
- array
- .iter()
- .filter_map(|m| m.as_ref())
- .map(|dvc| dvc.commit)
- .max()
- .unwrap_or(0)
+/// Why a `DoViewChange` body could not be read as a suffix.
+///
+/// Dropped whole rather than partially trusted: the merge indexes
arithmetically
+/// from the head op, so one bad offset misattributes a header, nack, or body.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum DvcSuffixError {
+ /// Body length is not a whole number of headers.
+ NotHeaderMultiple { body_len: usize },
+ /// More entries than the bitsets can address.
+ TooManyEntries { count: usize },
+ /// An entry is not a valid `PrepareHeader` bit pattern.
+ MalformedHeader { index: usize },
+ /// Entries are not consecutive descending from the head op.
+ OpOutOfOrder {
+ index: usize,
+ expected: u64,
+ found: u64,
+ },
+ /// A bitset addresses an entry the body does not contain.
+ BitsetBeyondSuffix { count: usize },
+ /// An entry's identity checksum does not match its own contents.
+ ChecksumMismatch { index: usize },
+ /// A lower entry claims a view newer than the entry above it.
+ ViewRegressed { index: usize },
+ /// A lower entry claims a timestamp at or after the entry above it.
+ TimestampNotDecreasing { index: usize },
+ /// Consecutive entries do not hash-chain.
+ ChainBreak { index: usize },
}
-/// Reset the DVC quorum array.
-pub const fn dvc_reset(array: &mut DvcQuorumArray) {
- *array = dvc_quorum_array_empty();
+impl std::fmt::Display for DvcSuffixError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::NotHeaderMultiple { body_len } => write!(
+ f,
+ "do_view_change body of {body_len} bytes is not a whole number
of {} -byte headers",
+ size_of::<PrepareHeader>()
+ ),
+ Self::TooManyEntries { count } => write!(
+ f,
+ "do_view_change suffix of {count} entries exceeds the maximum
{DVC_HEADERS_MAX}"
+ ),
+ Self::MalformedHeader { index } => {
+ write!(
+ f,
+ "do_view_change suffix entry {index} is not a prepare
header"
+ )
+ }
+ Self::OpOutOfOrder {
+ index,
+ expected,
+ found,
+ } => write!(
+ f,
+ "do_view_change suffix entry {index} carries op {found},
expected {expected}"
+ ),
+ Self::BitsetBeyondSuffix { count } => write!(
+ f,
+ "do_view_change bitset addresses an entry past the
{count}-entry suffix"
+ ),
+ Self::ChecksumMismatch { index } => write!(
+ f,
+ "do_view_change suffix entry {index} does not match its own
identity checksum"
+ ),
+ Self::ViewRegressed { index } => write!(
+ f,
+ "do_view_change suffix entry {index} claims a newer view than
the entry above it"
+ ),
+ Self::TimestampNotDecreasing { index } => write!(
+ f,
+ "do_view_change suffix entry {index} does not predate the
entry above it"
+ ),
+ Self::ChainBreak { index } => write!(
+ f,
+ "do_view_change suffix entry {index} does not chain to the
entry above it"
+ ),
+ }
+ }
}
-/// Iterator over all stored DVCs.
-// TODO: add #[must_use] -- pure iterator query, callers should not ignore.
-pub fn dvc_iter(array: &DvcQuorumArray) -> impl Iterator<Item = &StoredDvc> {
- array.iter().filter_map(|m| m.as_ref())
+impl std::error::Error for DvcSuffixError {}
+
+/// Read a suffix out of a `DoViewChange` body.
+///
+/// `head_op` is the sender's `header.op`; entries run consecutively down from
it,
+/// blanks included, so slot `i` is unambiguously op `head_op - i`. An empty
body
+/// yields an empty suffix, which is what a sender with nothing to describe
sends.
+///
+/// `body` must carry [`PrepareHeader`]'s alignment: the cast below is
checked, so an
+/// unaligned body reports every entry as [`DvcSuffixError::MalformedHeader`]
instead
+/// of what is actually wrong. Real frames clear this because the body starts
a whole
+/// number of 256-byte headers into an aligned buffer; the debug assert
catches a
+/// hand-built one.
+///
+/// # Errors
+/// [`DvcSuffixError`] when the body is not a consecutive run of valid prepare
+/// headers descending from `head_op`, or a bitset addresses a missing entry.
+pub fn dvc_suffix_decode(
+ body: &[u8],
+ head_op: u64,
+ nack_bitset: u128,
+ present_bitset: u128,
+) -> Result<DvcSuffix, DvcSuffixError> {
+ debug_assert!(
+ body.is_empty()
+ || body
+ .as_ptr()
+ .addr()
+ .is_multiple_of(align_of::<PrepareHeader>()),
+ "suffix body must be aligned for PrepareHeader"
+ );
+ let header_size = size_of::<PrepareHeader>();
+ if !body.len().is_multiple_of(header_size) {
+ return Err(DvcSuffixError::NotHeaderMultiple {
+ body_len: body.len(),
+ });
+ }
+ let count = body.len() / header_size;
+ if count > DVC_HEADERS_MAX {
+ return Err(DvcSuffixError::TooManyEntries { count });
+ }
+ if count < DVC_HEADERS_MAX {
+ let beyond = !((1u128 << count) - 1);
+ if nack_bitset & beyond != 0 || present_bitset & beyond != 0 {
+ return Err(DvcSuffixError::BitsetBeyondSuffix { count });
+ }
+ }
+
+ let mut headers = Vec::with_capacity(count);
+ // The entry above the current one, skipping blanks: high-to-low, so the
child.
+ let mut child: Option<PrepareHeader> = None;
+ for index in 0..count {
Review Comment:
none of these entries go through `PrepareHeader::validate` - the
reserved-must-be-zero rule (one of the two reasons stated on `validate` is
exactly the `dvc_blank` exact-equality classification) is never enforced at the
one place that indexes blanks. an entry with a dirty reserved byte and
`checksum == 0` also skips the identity recompute below, so it reads as a Valid
header, conflicts with every honest sender's, and the merge refuses to pick a
canonical header - the view change stalls and repeats on every escalation for
as long as the entry survives. the DVC body is sealed and verified in transit,
so the exposure is a crafted frame from an authenticated peer or a
corrupt-at-rest unsealed header re-sealed on send - but the rule exists for
exactly that class. same story on the repair door:
`RepairPrepareHeader::validate` checks the command only. both ingress points
want the full validate, or at least the reserved-zero check.
##########
core/consensus/src/impls.rs:
##########
@@ -1885,16 +2051,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
.borrow_mut()
.reset(TimeoutKind::DoViewChangeMessage);
- let current_op = self.sequencer.current_sequence();
- let action = VsrAction::SendDoViewChange {
- view: self.view.get(),
- target: self.primary_index(self.view.get()),
- log_view: self.log_view.get(),
- op: current_op,
- // commit_max clamped to op: see `handle_start_view_change`.
- commit: self.commit_max.get().min(current_op),
- namespace: self.namespace,
- };
+ // The same snapshot the first send used. A retransmit that re-derived
the
Review Comment:
"the same snapshot the first send used" - it isn't: `build_do_view_change`
calls `local_dvc_suffix()` fresh each time, and the shard refreshes the suffix
on every SVC/DVC handler and non-Normal tick. worse, on a tick-driven
retransmit the `(op, commit)` tag can have moved with no refresh in between,
and `local_dvc_suffix()` then returns an *empty* suffix - the retransmit
silently retracts every nack and body offer. harmless today only because
`dvc_record` drops duplicate senders. either pin the snapshot for real or fix
the comment.
##########
core/binary_protocol/src/consensus/header.rs:
##########
@@ -1023,10 +1297,52 @@ impl ConsensusHeader for DoViewChangeHeader {
"commit cannot exceed op".to_string(),
));
}
+ let suffix_len = self.suffix_len()?;
+ // Bits past the suffix describe entries never sent: unchecked, a peer
could
+ // smuggle a nack for an op the new primary would then truncate.
+ if suffix_len < DVC_HEADERS_MAX {
+ let beyond = !((1u128 << suffix_len) - 1);
+ if self.nack_bitset & beyond != 0 || self.present_bitset & beyond
!= 0 {
+ return Err(ConsensusError::InvalidField(format!(
+ "do_view_change: bitset bits set past the
{suffix_len}-entry suffix"
+ )));
+ }
+ }
Ok(())
}
}
+impl DoViewChangeHeader {
+ /// Number of `PrepareHeader`s in the body.
+ ///
+ /// Zero is valid and means "no suffix": a replica with nothing
uncommitted, or a
+ /// peer predating the suffix. Both contribute numbers only.
Review Comment:
"a peer predating the suffix" can't reach this decode anymore - the frame
seal rejects a pre-seal peer's frame before any field is read (this file's
module doc: replicas upgraded together, cluster down). same stale pre-seal-peer
claims on `StartViewHeader::incarnation`,
`RequestStartViewHeader::incarnation`, `StartViewHeader::suffix_len`,
`view_change_quorum`'s decode comment, and the shard's suffix-verify comments.
worth striking them all and pointing at the module doc - keeping zero-tolerance
wording only where it's real, the on-disk `CHECKSUM_UNSEALED` path (which also
covers partition-plane prepares, so that skip stays).
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -169,6 +202,41 @@ where
Ok(current_op)
}
+/// Compute a prepare's identity checksum: which prepare this is, independent
of
+/// which view re-sent it.
+///
+/// `view` is excluded from the covered bytes even though the frame checksum
covers
+/// it, because `restamp_prepare_view` rewrites `view` in place to clear the
+/// receiver's `header.view < view` fence. Covering it would give one logical
op a
+/// different checksum per replica, and the merge would read those as competing
+/// prepares nacking each other. Not merely a workaround either: two frames
+/// differing only in `view` ARE the same op, and everything that makes them
+/// genuinely different (client, request, timestamp, parent, operation, or the
body
+/// via `checksum_body`) stays covered.
+///
+/// `checksum_body` must already be set, since it is how the body reaches this
+/// value. A prepare left unsealed there gets an identity over its header
alone.
+#[must_use]
+pub fn prepare_identity_checksum(header: &PrepareHeader) -> u128 {
Review Comment:
one-line pass-through to `header.identity_checksum()` -
`seal_prepare_checksum` ten lines down and the tests are the only users. the
`restamp_prepare_view` rationale in the doc is the part worth keeping; it can
live on `identity_checksum` itself and the wrapper can go.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -89,6 +165,10 @@ pub async fn persist_offset_max(
/// files, so the commit-path reader must agree or a torn file turns every
/// later commit-apply into an error. Real I/O errors still propagate: mapping
/// them to `None` would silently rewind a valid higher offset.
+///
+/// A checksum mismatch is an error, not `None`. `None` means "no offset
recorded",
+/// which the caller folds as `max(absent, incoming)` and overwrites; doing
that to a
+/// failed-checksum file discards a cursor that may have been far ahead.
async fn read_persisted_offset(path: &str) -> Result<Option<u64>, IggyError> {
if !Path::new(path).exists() {
Review Comment:
blocking `Path::exists()` stat on the pump - and `persist_offset` above
documents removing exactly this pattern. it's also redundant: map `open`'s
`NotFound` to `Ok(None)` instead (keep that arm, or every cold key becomes a
hard error).
--
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]
