diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 33838ee636..3a83c0e2ce 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -3194,11 +3194,11 @@ impl> VsrConsensus { // 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 86170553d7..9856dd02ad 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 = + ::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 0725798a4d..762c932f20 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 = 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( shard: u16, partition: &mut IggyPartition, - 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 = 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( 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( } // 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( 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"