fix(metadata): truncate the WAL suffix above the state-transfer floor

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();
         }

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;

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"