This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 89dea0ed3 fix(metadata): truncate the WAL suffix above the
state-transfer floor (#3918)
89dea0ed3 is described below
commit 89dea0ed3c8853ba27b38b28dde5f7263779126b
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Thu Aug 20 14:15:08 2026 +0200
fix(metadata): truncate the WAL suffix above the state-transfer floor
(#3918)
---
core/consensus/src/impls.rs | 10 +--
core/metadata/src/impls/metadata.rs | 150 +++++++++++++++++++++++++++++++-----
core/shard/src/lib.rs | 58 +++++++++-----
3 files changed, 174 insertions(+), 44 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 33838ee63..3a83c0e2c 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -3194,11 +3194,11 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
// outranks the real primary's and pushes ops that view already
discarded back
// over committed bodies.
//
- // TODO(suffix-truncation): re-adoption drops the head again, but the
WAL
- // still holds the discarded suffix, so the primary's next prepare
lands on an
- // op that suffix already occupies and fails `append`'s slot-collision
check,
- // poisoning the journal. Loud beats the silent divergence above; a
durable
- // truncate-from-op primitive is what actually closes it.
+ // Re-adoption drops the head again while the WAL still holds the
discarded
+ // suffix. Consensus is sans-io and cannot truncate it; the plane
sweeps it
+ // on every adoption (`reconcile_{metadata,partition}_view_divergence`,
+ // above-head branch) before the primary's next prepare can collide
with a
+ // relic in `append`'s slot-collision check.
if msg_view == self.log_view.get() && msg_op < self.commit_min() {
return Vec::new();
}
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index 86170553d..9856dd02a 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -1694,6 +1694,40 @@ where
let local_applied = consensus.commit_min();
let snapshot_ahead = snapshot_seq > local_applied;
+ if snapshot_ahead && let Some(journal) = &self.journal {
+ // Discard the WAL suffix above the incoming floor BEFORE anything
+ // installs: the commit walk matches entries by op number alone, so
+ // a pre-crash suffix a view change has since reassigned would be
+ // applied as committed once the floor jump below pulls the walk
+ // past it. Ahead of the snapshot restore so a failed truncate
+ // aborts a not-yet-started install (the transfer retries) instead
+ // of stranding a restored state machine without its floor jump,
+ // which would double-apply the snapshot's ops on the next walk.
+ // Committed ops the range covered come back through the
+ // post-install tail repair; uncommitted ones were decided away by
+ // the view change that made this replica a transfer receiver.
+ // Serialization against appends holds as in
+ // `reconcile_metadata_view_divergence`: the pump is
+ // single-threaded and `replicate_preflight` refuses prepares
+ // while `is_transferring`.
+ let removed = journal
+ .handle()
+ .truncate_from(snapshot_seq + 1)
+ .await
+ .map_err(SnapshotError::Io)?;
+ if removed > 0 {
+ // The DVC snapshot's `(op, commit)` tag does not move when
+ // entries are removed under it; left stale it would advertise
+ // headers this replica can no longer serve.
+ consensus.invalidate_local_dvc_suffix();
+ tracing::warn!(
+ snapshot_seq,
+ removed,
+ "state transfer dropped {removed} journal entries above
the incoming floor"
+ );
+ }
+ }
+
if snapshot_ahead {
if let Some(coordinator) = &self.coordinator {
// The transferred snapshot REPLACES the one the superblock's
@@ -1760,26 +1794,9 @@ where
// The snapshot IS ops `..=snapshot_seq` applied: jump the applied
// frontier (this is the op-jump the tail repair resumes from) and
// let the announced commit point pull the walk target forward.
- //
- // TODO(suffix-truncation): this hands the commit walk a floor it
never
- // verified. `set_snapshot_op` above only marks entries at or
below the
- // floor EVICTABLE -- everything above it stays resident -- and
the walk
- // matches WAL entries by op number alone, with no view or
hash-chain
- // check. A node carrying a pre-crash prepared-but-uncommitted
suffix that
- // a view change has since reassigned cluster-side will therefore
apply
- // those stale bodies as committed. The client table is shielded
(the
- // `client_table_frontier` fence skips table effects at or below
the
- // transferred frontier); the state machine is not.
- //
- // Same root cause as the `TODO(suffix-truncation)` in
- // `VsrConsensus::handle_start_view`, and the same missing piece
closes
- // both: a durable truncate-from-op primitive on the journal, so a
floor
- // jump can discard the suffix above it instead of leaving it to be
- // matched by op number. The floor jump does not create the hole,
but it
- // widens exposure to it precisely on the nodes guaranteed to have
a stale
- // log -- every state-transfer receiver is one. Deliberately out
of scope
- // here (flagged in review as follow-up): the primitive is a
journal
- // durability change, not a state-transfer one.
+ // The walk matches WAL entries by op number alone, so this floor
+ // is only safe because the truncate at the top of this install
+ // already discarded every journal entry above it.
consensus.set_commit_floor(snapshot_seq);
if snapshot_seq > consensus.sequencer().current_sequence() {
consensus.sequencer().set_sequence(snapshot_seq);
@@ -5297,6 +5314,99 @@ mod tests {
/// `resume_stranded_commits` (wired into the shard pump tick) is the
/// backstop: it re-enters the commit path, applies the stranded prefix,
/// and promotes the queued register, whose awaiter then resolves.
+ /// A state-transfer receiver's WAL can hold a pre-crash suffix above the
+ /// incoming floor. The commit walk matches entries by op number alone, so
+ /// installing without discarding that suffix would later apply its stale
+ /// bodies as committed. The install must truncate the WAL above
+ /// `snapshot_seq` and keep everything at or below it, which tail repair
+ /// resumes from.
+ #[compio::test]
+ async fn
state_transfer_install_truncates_the_wal_above_the_incoming_floor() {
+ const CLIENT: u128 = 9;
+ const SESSION: u64 = 1;
+ const ACTING_USER: u32 = 7;
+ const SNAPSHOT_SEQ: u64 = 2;
+
+ let dir = tempfile::tempdir().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_GROUP,
+ NoopBus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (),
TestMux> =
+ IggyMetadata::new(
+ Some(consensus),
+ Some(journal),
+ None,
+ None,
+ TestMux::default(),
+ None,
+ );
+ let consensus = md.consensus.as_ref().unwrap();
+ md.client_table.borrow_mut().commit_register(
+ CLIENT,
+ ACTING_USER,
+ register_reply(CLIENT, SESSION),
+ );
+
+ // Journal ops 1..=3 directly (no acks, so `commit_min` stays 0 and the
+ // transfer is "ahead"): 1 and 2 sit at or below the incoming floor, 3
is
+ // the relic suffix a view change has since reassigned.
+ for (request_id, name) in [(1, "s1"), (2, "s2"), (3, "s3")] {
+ let projected = md
+ .prepare_request(create_stream_request(CLIENT, request_id,
name))
+ .expect("CreateStream is client-allowed");
+ consensus.pipeline_message(PlaneKind::Metadata, &projected);
+ md.journal
+ .as_ref()
+ .unwrap()
+ .handle()
+ .append(projected)
+ .await
+ .expect("seeding the WAL succeeds");
+ }
+ let journal_handle = md.journal.as_ref().unwrap().handle();
+ assert_eq!(journal_handle.last_op(), Some(3));
+
+ // A donor mux fills the snapshot the way a serving primary would, so
+ // the restore path sees populated sections rather than a bare
envelope.
+ let snapshot_bytes =
+ <IggySnapshot as Snapshot>::create(&TestMux::default(),
SNAPSHOT_SEQ, 1)
+ .expect("donor snapshot builds")
+ .encode()
+ .expect("donor snapshot encodes");
+ md.install_state_transfer(
+ &snapshot_bytes,
+ ClientTable::new(CLIENTS_TABLE_MAX),
+ 0,
+ SNAPSHOT_SEQ,
+ )
+ .await
+ .expect("install succeeds");
+
+ assert_eq!(
+ journal_handle.last_op(),
+ Some(SNAPSHOT_SEQ),
+ "the relic above the incoming floor must be gone"
+ );
+ assert!(
+ journal_handle.header(3).is_none(),
+ "op 3 was above the floor; the commit walk must never see it again"
+ );
+ assert!(
+ journal_handle.header(1).is_some() &&
journal_handle.header(2).is_some(),
+ "ops at or below the floor stay for the walk and tail repair"
+ );
+ }
+
#[compio::test]
async fn tick_backstop_must_resume_stranded_commits_and_promotions() {
use std::future::Future;
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 0725798a4..762c932f2 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3882,10 +3882,13 @@ where
.consensus()
.handle_start_view(PlaneKind::Partitions, &header,
suffix_body);
let adopted = !actions.is_empty();
- if adopted && let Some(pending) =
partition.consensus().pending_view_log() {
+ if adopted {
// Ahead of the local dispatch, which rebuilds the pipeline out of
the
- // journal this rewrites. Same position as the metadata arm's twin.
- reconcile_partition_view_divergence(self.id, partition,
&pending).await;
+ // journal this rewrites. Same position as the metadata arm's
twin, and
+ // like it, pending-less adoptions (empty StartView suffix) still
sweep
+ // the relics above the adopted head.
+ let pending = partition.consensus().pending_view_log();
+ reconcile_partition_view_divergence(self.id, partition,
pending.as_ref()).await;
}
let consensus = partition.consensus();
let (local_actions, wire_actions) = split_local_actions(actions);
@@ -4895,7 +4898,7 @@ where
// already holds a header for, so the scan would report a gap
nothing
// fills. Backups reach this on StartView adoption; a
primary-elect has no
// adoption to hang it off.
- reconcile_partition_view_divergence(self.id, partition,
&pending).await;
+ reconcile_partition_view_divergence(self.id, partition,
Some(&pending)).await;
let consensus = partition.consensus();
// Identity, not presence: see the metadata twin. The floor is the
local
// commit point, the partition twin of the metadata snapshot floor:
@@ -5030,6 +5033,13 @@ where
/// `RebuildPipeline` reads the pipeline back out of it, and
`CommitJournal`
/// applies whatever sits at each op up to the merged commit point.
///
+ /// Runs on every adoption, parked suffix or not: an EMPTY `StartView`
suffix
+ /// (`commit == op`, the steady case) parks no pending log, yet adoption
still
+ /// drops the head under any journaled relics above it, and the primary's
next
+ /// prepare would collide with them in `append` and poison the journal.
With
+ /// no pending log the divergence scan has nothing to walk and only the
+ /// above-head sweep applies, with the head read off the adopted sequencer.
+ ///
/// 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
@@ -5052,9 +5062,7 @@ where
let Some(ref consensus) = metadata.consensus else {
return;
};
- let Some(pending) = consensus.pending_view_log() else {
- return;
- };
+ let pending = consensus.pending_view_log();
let Some(journal) = metadata.journal.as_ref() else {
return;
};
@@ -5063,10 +5071,11 @@ where
// 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 announced_commit = pending.as_ref().map_or(0, |pending|
pending.commit_max);
+ let applied_floor = announced_commit.max(consensus.commit_min());
let mut repairable_from: Option<u64> = None;
- for canonical in &pending.headers {
+ for canonical in pending.as_ref().map_or(&[][..], |pending|
&pending.headers) {
let Some(local) = usize::try_from(canonical.op)
.ok()
.and_then(|slot| journal.handle().header(slot))
@@ -5081,7 +5090,7 @@ where
shard = self.id,
op = canonical.op,
view = consensus.view(),
- commit_max = pending.commit_max,
+ commit_max = announced_commit,
commit_min = consensus.commit_min(),
local_checksum = local.checksum,
canonical_checksum = canonical.checksum,
@@ -5100,7 +5109,12 @@ where
// under it, and the next prepare at `op_head + 1` then collides in
`append`,
// which refuses the slot even when the ops match. Floored at the
applied
// point too: an executed op is not rollback-able whatever the head
says.
- let above_head = pending.op_head.max(applied_floor) + 1;
+ // With no parked suffix the adopted sequencer IS the announced head.
+ let op_head = pending.as_ref().map_or_else(
+ || consensus.sequencer().current_sequence(),
+ |pending| pending.op_head,
+ );
+ let above_head = op_head.max(applied_floor) + 1;
if journal
.handle()
.last_op()
@@ -5128,7 +5142,7 @@ where
shard = self.id,
from_op,
removed,
- op_head = pending.op_head,
+ 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"
@@ -8766,17 +8780,18 @@ fn build_dvc_suffix(
async fn reconcile_partition_view_divergence<B, SB>(
shard: u16,
partition: &mut IggyPartition<B, SB>,
- pending: &MergedLog,
+ pending: Option<&MergedLog>,
) where
B: MessageBus,
SB: journal::superblock::SuperblockStore,
{
// Truncation is safe only above what this replica has *applied*, which is
not
// the view's commit point: a backup can sit above it.
- let applied_floor =
pending.commit_max.max(partition.consensus().commit_min());
+ let announced_commit = pending.map_or(0, |pending| pending.commit_max);
+ let applied_floor =
announced_commit.max(partition.consensus().commit_min());
let mut repairable_from: Option<u64> = None;
- for canonical in &pending.headers {
+ for canonical in pending.map_or(&[][..], |pending| &pending.headers) {
let Some(local) =
partition.log.journal().inner.header_by_op(canonical.op) else {
continue;
};
@@ -8789,7 +8804,7 @@ async fn reconcile_partition_view_divergence<B, SB>(
namespace_raw = partition.consensus().group(),
op = canonical.op,
view = partition.consensus().view(),
- commit_max = pending.commit_max,
+ commit_max = announced_commit,
commit_min = partition.consensus().commit_min(),
local_checksum = local.checksum,
canonical_checksum = canonical.checksum,
@@ -8803,8 +8818,13 @@ async fn reconcile_partition_view_divergence<B, SB>(
}
// The suffix above the announced head, which no canonical header names.
As on
- // the metadata twin, except here `append` pushes a duplicate rather than
erroring.
- let above_head = pending.op_head.max(applied_floor) + 1;
+ // the metadata twin, except here `append` pushes a duplicate rather than
+ // erroring. With no parked suffix the adopted sequencer IS the announced
head.
+ let op_head = pending.map_or_else(
+ || partition.consensus().sequencer().current_sequence(),
+ |pending| pending.op_head,
+ );
+ let above_head = op_head.max(applied_floor) + 1;
if partition
.log
.journal()
@@ -8825,7 +8845,7 @@ async fn reconcile_partition_view_divergence<B, SB>(
namespace_raw = partition.consensus().group(),
from_op,
removed,
- op_head = pending.op_head,
+ op_head,
view = partition.consensus().view(),
"dropped {removed} uncommitted partition entries from op
{from_op} that \
disagreed with the view's log; the primary's retransmission
refills the range"
