This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch vsr-dvc-headers
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/vsr-dvc-headers by this push:
new e2819a567 fix: address review comment
e2819a567 is described below
commit e2819a56776caaa084d7a24dc766f016c10f0c2c
Author: Krishna Vishal <[email protected]>
AuthorDate: Thu Aug 6 15:09:16 2026 +0530
fix: address review comment
---
core/consensus/src/dvc_merge.rs | 45 ++++++++++
.../tests/cluster/metadata_checkpoint_restart.rs | 6 +-
core/metadata/src/impls/metadata.rs | 99 +++++++++++++++++++++-
core/partitions/src/iggy_partitions.rs | 83 ++++++++++++++++++
core/partitions/src/journal.rs | 25 ++++++
core/shard/src/lib.rs | 44 +++++++++-
6 files changed, 296 insertions(+), 6 deletions(-)
diff --git a/core/consensus/src/dvc_merge.rs b/core/consensus/src/dvc_merge.rs
index 79331a80c..bbf265248 100644
--- a/core/consensus/src/dvc_merge.rs
+++ b/core/consensus/src/dvc_merge.rs
@@ -619,6 +619,51 @@ mod tests {
);
}
+ #[test]
+ fn given_blank_commit_point_from_every_sender_should_deadlock() {
+ // The commit point is scanned and may not be discarded, so a sender
that
+ // reports it blank is deferring to a peer. When every sender defers
there
+ // is no peer left and the view cannot start.
+ //
+ // Nothing in the merge can rescue this, which is why the senders must
not
+ // produce it: a replica keeps the header at its own commit point
through
+ // compaction (the metadata checkpoint drain stops one op short, a
+ // partition answers from its evicted ring).
+ let mut quorum = dvc_quorum_array_empty();
+ let blank_at_commit = DvcSuffix::new(vec![dvc_blank(5)], 0, 0);
+ for replica in 0..3 {
+ dvc_record(&mut quorum, dvc(replica, 1, 5, 5,
blank_at_commit.clone()));
+ }
+
+ assert_eq!(
+ merge_dvc_quorum(&quorum, quorums_r3()),
+ MergeOutcome::Deadlocked { undecided_op: 5 },
+ "a blank commit point is neither adoptable nor discardable"
+ );
+ }
+
+ #[test]
+ fn given_blank_commit_point_from_one_sender_should_adopt_the_peer_header()
{
+ // The same suffix stops being fatal the moment one sender still holds
the
+ // header: that one is canonical and serves the body, and the deferring
+ // sender neither nacks it nor conflicts with it.
+ let mut quorum = dvc_quorum_array_empty();
+ dvc_record(
+ &mut quorum,
+ dvc(0, 1, 5, 5, DvcSuffix::new(vec![dvc_blank(5)], 0, 0)),
+ );
+ dvc_record(
+ &mut quorum,
+ dvc(1, 1, 5, 5, suffix_all_present(suffix_headers(5, 5, 1))),
+ );
+
+ let MergeOutcome::Ready(log) = merge_dvc_quorum(&quorum, quorums_r3())
else {
+ panic!("one surviving copy of the commit point is enough to start
the view");
+ };
+ assert_eq!(log.op_head, 5);
+ assert_eq!(log.commit_max, 5);
+ }
+
#[test]
fn
given_header_without_a_servable_body_when_replicas_outstanding_should_await_repair()
{
// Both senders have op 4's header, neither can serve its body, and
replica 2
diff --git a/core/integration/tests/cluster/metadata_checkpoint_restart.rs
b/core/integration/tests/cluster/metadata_checkpoint_restart.rs
index 75ae60f6e..d3d9a70e8 100644
--- a/core/integration/tests/cluster/metadata_checkpoint_restart.rs
+++ b/core/integration/tests/cluster/metadata_checkpoint_restart.rs
@@ -119,8 +119,10 @@ async fn await_checkpoint_on_all_nodes(harness:
&TestHarness, generation: usize)
// a checkpoint, so the transfer descriptor's `commit_op == snapshot_seq` and
// the post-install tail repair has nothing to fetch (`commit_min ==
// commit_max` skips it). The below-floor retry then proves the reply ring
-// rode the transferred table: request 191's reply was minted at op 192, which
-// every node drained out of its WAL at that same checkpoint.
+// rode the transferred table: request 191's reply was minted at op 192, and
+// replay starts at `snapshot_seq + 1`, so no node re-executes it. The
+// checkpoint drain keeps op 192's entry as the commit-point header a
+// `DoViewChange` needs, but never replays it.
#[iggy_harness(cluster_nodes = 3, server(metadata.journal_slots = "256"))]
async fn given_drained_journal_when_node_restarts_should_install_snapshot_only(
harness: &mut TestHarness,
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index e523c99a9..4bf974cd1 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -423,17 +423,29 @@ impl<M> SnapshotCoordinator<M> {
Ok(checksum)
}
- /// Drain the snapshotted prefix `0..=last_op` to reclaim WAL space. Runs
only
- /// after the pairing is durable (see [`Self::persist_snapshot`]).
+ /// Drain the snapshotted prefix below `last_op` to reclaim WAL space. Runs
+ /// only after the pairing is durable (see [`Self::persist_snapshot`]).
+ ///
+ /// `last_op` itself is retained, one entry the snapshot has already
+ /// superseded. It is this replica's commit point, and a `DoViewChange`
+ /// carries a header for every op from there up. Draining it inclusively
+ /// leaves that entry blank, and blank at the commit point is the one slot
+ /// the merge can neither adopt nor discard: a quorum of senders that all
+ /// checkpointed at the same op deadlocks the view change
+ /// (`dvc_merge::merge_dvc_quorum`). Reclaiming one more entry is not worth
+ /// a group that cannot elect.
#[allow(clippy::future_not_send)]
async fn drain<J: JournalHandle>(
&self,
journal: &J,
last_op: u64,
) -> Result<(), SnapshotError> {
+ let Some(drain_to) = last_op.checked_sub(1) else {
+ return Ok(());
+ };
journal
.handle()
- .drain(0..=last_op)
+ .drain(0..=drain_to)
.await
.map_err(SnapshotError::Io)?;
Ok(())
@@ -4808,6 +4820,87 @@ mod tests {
);
}
+ /// A checkpoint reclaims the WAL prefix the snapshot supersedes, but must
+ /// stop one op short of the checkpoint op itself.
+ ///
+ /// That op is the replica's commit point, and its `DoViewChange` suffix is
+ /// floored there. The merge scans the commit point and may not discard it,
+ /// so a sender with no header to put there is deferring to a peer; when
+ /// every sender has checkpointed at the same op the view change deadlocks
+ /// (`dvc_merge::merge_dvc_quorum`). Checkpoints fire on local journal
+ /// occupancy, which is symmetric across replicas seeing the same ops, so
+ /// "every sender" is the ordinary case, not a coincidence.
+ #[compio::test]
+ async fn checkpoint_drain_retains_the_commit_point_header() {
+ const CLIENT: u128 = 1;
+ const SESSION: u64 = 1;
+ const ACTING_USER: u32 = 7;
+ const OPS: u64 = 5;
+ const CHECKPOINT_OP: u64 = 3;
+
+ let dir = tempfile::tempdir().unwrap();
+
std::fs::create_dir_all(dir.path().join(crate::impls::METADATA_DIR)).unwrap();
+ let journal =
+
journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"),
0)
+ .await
+ .unwrap();
+ let consensus = VsrConsensus::new(
+ 1,
+ 0,
+ 1,
+ server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ NoopBus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (),
TestMux> =
+ IggyMetadata::new(
+ Some(consensus),
+ Some(journal),
+ None,
+ None,
+ TestMux::default(),
+ Some(dir.path().to_path_buf()),
+ );
+ let consensus = md.consensus.as_ref().unwrap();
+ md.client_table.borrow_mut().commit_register(
+ CLIENT,
+ ACTING_USER,
+ register_reply(CLIENT, SESSION),
+ );
+
+ for op in 1..=OPS {
+ let prepare = md
+ .prepare_request(create_stream_request(CLIENT, op,
&format!("s{op}")))
+ .expect("CreateStream is client-allowed");
+ consensus.pipeline_message(PlaneKind::Metadata, &prepare);
+ md.on_replicate(prepare).await;
+ }
+
+ let journal = md.journal.as_ref().unwrap();
+ md.coordinator
+ .as_ref()
+ .expect("data_dir present arms the coordinator")
+ .drain(journal, CHECKPOINT_OP)
+ .await
+ .expect("drain the snapshotted prefix");
+
+ let header_at = |op: u64|
journal.header(usize::try_from(op).expect("test ops fit usize"));
+ for op in 1..CHECKPOINT_OP {
+ assert!(
+ header_at(op).is_none(),
+ "op {op} is below the checkpoint and must be reclaimed"
+ );
+ }
+ assert!(
+ header_at(CHECKPOINT_OP).is_some(),
+ "the checkpoint op is the commit point and must stay describable
in a DVC"
+ );
+ for op in CHECKPOINT_OP + 1..=OPS {
+ assert!(header_at(op).is_some(), "op {op} was never snapshotted");
+ }
+ }
+
/// Reproduces the single-node "metadata prepare queue is full" wedge
///
/// `checkpoint_if_needed` runs inside `on_replicate`, once per submit.
diff --git a/core/partitions/src/iggy_partitions.rs
b/core/partitions/src/iggy_partitions.rs
index 5377701a1..48458f2f2 100644
--- a/core/partitions/src/iggy_partitions.rs
+++ b/core/partitions/src/iggy_partitions.rs
@@ -652,6 +652,28 @@ mod tests {
)
}
+ /// `build_partition` for a replicated group. The replica count is what
+ /// decides whether the journal retains evicted entries for repair, so a
+ /// single-replica partition cannot exercise anything that reads the ring.
+ fn build_replicated_partition() -> IggyPartition<IggyMessageBus> {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let consensus = VsrConsensus::new(
+ TEST_CLUSTER,
+ 0,
+ 3,
+ namespace.inner(),
+ IggyMessageBus::new(0),
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ IggyPartition::with_in_memory_storage(
+ Arc::new(PartitionStats::default()),
+ consensus,
+ IggyByteSize::from(1024 * 1024),
+ false,
+ )
+ }
+
/// One-message `SendMessages` journal entry stamped at `op` /
`base_offset`.
/// Reuses the production blob builder + checksum stamping so the entry
/// decodes through `decode_prepare_slice` and indexes into `offset_to_op`,
@@ -794,6 +816,67 @@ mod tests {
);
}
+ /// A flush evicts the committed prefix up to and INCLUDING `commit_max`,
so
+ /// a caught-up replica keeps no resident header at its own commit point.
The
+ /// `DoViewChange` suffix is floored there and cannot nack it, so reading
the
+ /// resident headers alone sends the commit point out blank, which a
quorum of
+ /// senders turns into a view change that never starts.
+ ///
+ /// The entry is still servable (`repair_entry` answers from the evicted
+ /// ring), so the suffix reads through `repair_header`, over the same
range.
+ #[compio::test]
+ async fn evicted_commit_point_still_answers_for_the_view_change_suffix() {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let partition = build_replicated_partition();
+
+ for offset in 0..=2u64 {
+ partition
+ .log
+ .journal()
+ .inner
+ .append(build_send_messages_entry(namespace, offset + 1,
offset))
+ .await
+ .expect("append journal entry");
+ }
+
+ let commit_max = 3;
+ let prefix =
partition.log.journal().inner.committed_prefix(commit_max);
+ assert_eq!(prefix.len(), 3, "the whole log is committed and
flushable");
+ partition
+ .log
+ .journal()
+ .inner
+ .evict_prefix(prefix.len())
+ .await;
+
+ assert!(
+ partition
+ .log
+ .journal()
+ .inner
+ .header_by_op(commit_max)
+ .is_none(),
+ "the flush evicted the commit point from the resident headers",
+ );
+ assert!(
+ partition
+ .log
+ .journal()
+ .inner
+ .repair_entry(commit_max)
+ .is_some(),
+ "yet the entry is still servable from the evicted ring",
+ );
+
+ let header = partition
+ .log
+ .journal()
+ .inner
+ .repair_header(commit_max)
+ .expect("the commit point must stay describable for the DVC
suffix");
+ assert_eq!(header.op, commit_max);
+ }
+
/// The resident journal holds replicated-but-uncommitted prepares ahead of
/// the commit frontier. A poll must clamp at `ceiling` (the commit offset)
/// so it never returns a dirty read of view-change-rollbackable data, even
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index 0b2eaf8d0..cf3988777 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -325,6 +325,31 @@ impl PartitionJournal<PartitionJournalMemStorage> {
.map(|(_, entry)| entry.clone())
}
+ /// The header at `op`, over exactly the range [`Self::repair_entry`]
serves.
+ ///
+ /// NOT [`Self::header_by_op`], which reads the resident headers alone. The
+ /// committed prefix is evicted from those the moment its bytes reach a
+ /// segment, up to and including `commit_max`, so a `DoViewChange` built
off
+ /// the resident headers reports its own commit point blank. The merge
scans
+ /// the commit point and cannot discard it, so a quorum of such senders is
+ /// undecidable and the view never starts (`dvc_merge::merge_dvc_quorum`).
+ /// The entry is still servable from the evicted ring, which is what makes
+ /// the blank wrong rather than merely pessimistic.
+ ///
+ /// The ring drops from the front, so the highest evicted op -- the commit
+ /// point of the last flush -- is the last thing it forgets.
+ pub fn repair_header(&self, op: u64) -> Option<PrepareHeader> {
+ if let Some(header) = self.header_by_op(op) {
+ return Some(header);
+ }
+ let ring = unsafe { &*self.evicted_ring.get() };
+ let (_, entry) = ring.iter().find(|(ring_op, _)| *ring_op == op)?;
+ let header_bytes = entry.as_slice().get(..PREPARE_HEADER_SIZE)?;
+ bytemuck::checked::try_from_bytes::<PrepareHeader>(header_bytes)
+ .ok()
+ .copied()
+ }
+
/// Oldest op this journal can still serve for repair (ring front, else
/// resident head), or `None` when it holds nothing at all.
pub fn repair_retained_from(&self) -> Option<u64> {
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index f75d4dc81..9c0e010a8 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -5642,6 +5642,10 @@ where
/// 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>(partition: &partitions::IggyPartition<B>)
where
B: MessageBus,
@@ -5657,7 +5661,7 @@ where
let suffix = build_dvc_suffix(
commit,
op,
- |entry_op| journal.inner.header_by_op(entry_op),
+ |entry_op| journal.inner.repair_header(entry_op),
pending.as_ref().map(|pending| pending.headers.as_slice()),
);
consensus.set_local_dvc_suffix(suffix);
@@ -5881,6 +5885,25 @@ fn build_dvc_suffix(
headers.push(dvc_blank(entry_op));
if entry_op > commit {
nack_bitset |= 1u128 << index;
+ } else {
+ // The commit point, the one slot that goes out blank AND
+ // un-nacked. The merge scans it and may not discard it, so a
+ // sender is asking the new primary to take the header from
+ // someone else; if every sender in the quorum does that, the
+ // op is undecidable and the view never starts.
+ //
+ // Every compaction path is supposed to leave this header
behind
+ // (the metadata checkpoint drain stops one op short, a
+ // partition serves it from the evicted ring), so reaching here
+ // means a replica whose log genuinely starts above its own
+ // commit point: a state-transfer receiver that jumped its
+ // commit floor to a snapshot whose prepares it never held.
+ tracing::warn!(
+ op = entry_op,
+ commit,
+ "no header at this replica's commit point; the DVC reports
it blank and \
+ cannot nack it, so the view change stalls unless a peer
supplies it"
+ );
}
}
}
@@ -6497,6 +6520,25 @@ mod dvc_suffix_window_tests {
assert_eq!(suffix.nack_bitset(), 0b010, "only the blank op nacks");
}
+ #[test]
+ fn
given_no_header_at_the_commit_point_should_report_it_blank_and_undecidable() {
+ // The window's floor is the commit point, and a blank there is the one
+ // entry that goes out with neither a header nor a nack. The merge
scans
+ // that op and may not discard it, so a quorum of these deadlocks the
view
+ // change. Pinned here because both compaction paths are meant to keep
the
+ // header alive precisely so this shape never leaves a healthy replica.
+ let suffix = build_dvc_suffix(5, 5, |_| None, None);
+
+ assert_eq!(suffix.len(), 1);
+ assert_eq!(floor(&suffix), Some(5));
+ assert_eq!(
+ suffix.nack_bitset(),
+ 0,
+ "the commit point is never nacked, whatever the journal says"
+ );
+ assert_eq!(suffix.present_bitset(), 0);
+ }
+
#[test]
fn
given_a_window_at_the_depth_ceiling_when_building_should_floor_at_the_commit() {
// At the deepest legal prepare-queue depth the window still starts
exactly
