kvstorage: assert ReplicaID on destruction

DestroyReplica already performs a read of RaftReplicaID, so we can
assert on it at no additional cost.

Epic: none
Release note: none

Author: pav-kv

pkg/kv/kvserver/kvstorage/destroy.go

@@ -118,28 +118,30 @@ const DestroyReplicaTODO = 0
 // not be cleared.
 func DestroyReplica(
 	ctx context.Context,
-	rangeID roachpb.RangeID,
+	id roachpb.FullReplicaID,
 	reader storage.Reader,
 	writer storage.Writer,
 	nextReplicaID roachpb.ReplicaID,
 	opts ClearRangeDataOptions,
 ) error {
-	diskReplicaID, err := stateloader.Make(rangeID).LoadRaftReplicaID(ctx, reader)
+	diskReplicaID, err := stateloader.Make(id.RangeID).LoadRaftReplicaID(ctx, reader)
 	if err != nil {
 		return err
 	}
-	if diskReplicaID.ReplicaID >= nextReplicaID {
-		return errors.AssertionFailedf("replica r%d/%d must not survive its own tombstone", rangeID, diskReplicaID)
+	if repID := diskReplicaID.ReplicaID; repID != id.ReplicaID {
+		return errors.AssertionFailedf("replica %v has a mismatching ID %d", id, repID)
+	} else if repID >= nextReplicaID {
+		return errors.AssertionFailedf("replica %v must not survive its own tombstone", id)
 	}
 	_ = DestroyReplicaTODO // 2.1 + 2.2 + 3.1
-	if err := ClearRangeData(ctx, rangeID, reader, writer, opts); err != nil {
+	if err := ClearRangeData(ctx, id.RangeID, reader, writer, opts); err != nil {
 		return err
 	}
 
 	// Save a tombstone to ensure that replica IDs never get reused. Assert that
 	// the provided tombstone moves the existing one strictly forward. Failure to
 	// do so indicates that something is going wrong in the replica lifecycle.
-	sl := stateloader.Make(rangeID)
+	sl := stateloader.Make(id.RangeID)
 	ts, err := sl.LoadRangeTombstone(ctx, reader)
 	if err != nil {
 		return err

pkg/kv/kvserver/replica_app_batch.go

@@ -364,7 +364,7 @@ func (b *replicaAppBatch) runPostAddTriggersReplicaOnly(
 		// required for correctness, since the merge protocol should guarantee that
 		// no new replicas of the RHS can ever be created, but it doesn't hurt to
 		// be careful.
-		if err := kvstorage.DestroyReplica(ctx, rhsRepl.RangeID, b.batch, b.batch, mergedTombstoneReplicaID, kvstorage.ClearRangeDataOptions{
+		if err := kvstorage.DestroyReplica(ctx, rhsRepl.ID(), b.batch, b.batch, mergedTombstoneReplicaID, kvstorage.ClearRangeDataOptions{
 			ClearReplicatedByRangeID:   true,
 			ClearUnreplicatedByRangeID: true,
 		}); err != nil {
@@ -455,7 +455,7 @@ func (b *replicaAppBatch) runPostAddTriggersReplicaOnly(
 		// We've set the replica's in-mem status to reflect the pending destruction
 		// above, and DestroyReplica will also add a range tombstone to the
 		// batch, so that when we commit it, the removal is finalized.
-		if err := kvstorage.DestroyReplica(ctx, b.r.RangeID, b.batch, b.batch, change.NextReplicaID(), kvstorage.ClearRangeDataOptions{
+		if err := kvstorage.DestroyReplica(ctx, b.r.ID(), b.batch, b.batch, change.NextReplicaID(), kvstorage.ClearRangeDataOptions{
 			ClearReplicatedBySpan:      span,
 			ClearReplicatedByRangeID:   true,
 			ClearUnreplicatedByRangeID: true,

pkg/kv/kvserver/replica_destroy.go

@@ -111,7 +111,7 @@ func (r *Replica) destroyRaftMuLocked(ctx context.Context, nextReplicaID roachpb
 		ClearUnreplicatedByRangeID: true,
 	}
 	// TODO(sep-raft-log): need both engines separately here.
-	if err := kvstorage.DestroyReplica(ctx, r.RangeID, r.store.TODOEngine(), batch, nextReplicaID, opts); err != nil {
+	if err := kvstorage.DestroyReplica(ctx, r.ID(), r.store.TODOEngine(), batch, nextReplicaID, opts); err != nil {
 		return err
 	}
 	preTime := timeutil.Now()

pkg/kv/kvserver/replica_raftstorage.go

@@ -566,7 +566,7 @@ func (r *Replica) applySnapshotRaftMuLocked(
 		Term:  kvpb.RaftTerm(nonemptySnap.Metadata.Term),
 	}
 
-	subsumedDescs := make([]*roachpb.RangeDescriptor, 0, len(subsumedRepls))
+	subsume := make([]destroyReplicaInfo, 0, len(subsumedRepls))
 	for _, sr := range subsumedRepls {
 		// We mark the replica as destroyed so that new commands are not
 		// accepted. This destroy status will be detected after the batch
@@ -580,10 +580,11 @@ func (r *Replica) applySnapshotRaftMuLocked(
 		sr.shMu.destroyStatus.Set(
 			kvpb.NewRangeNotFoundError(sr.RangeID, sr.store.StoreID()),
 			destroyReasonRemoved)
+		srDesc := sr.descRLocked()
 		sr.mu.Unlock()
 		sr.readOnlyCmdMu.Unlock()
 
-		subsumedDescs = append(subsumedDescs, sr.Desc())
+		subsume = append(subsume, destroyReplicaInfo{id: sr.ID(), desc: srDesc})
 	}
 
 	sb := snapWriteBuilder{
@@ -593,10 +594,10 @@ func (r *Replica) applySnapshotRaftMuLocked(
 		sl:       r.raftMu.stateLoader,
 		writeSST: inSnap.SSTStorageScratch.WriteSST,
 
-		truncState:    truncState,
-		hardState:     hs,
-		desc:          desc,
-		subsumedDescs: subsumedDescs,
+		truncState: truncState,
+		hardState:  hs,
+		desc:       desc,
+		subsume:    subsume,
 
 		cleared: inSnap.clearedSpans,
 	}

pkg/kv/kvserver/snapshot_apply_prepare.go

@@ -20,6 +20,14 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
+// destroyReplicaInfo contains the replica's metadata needed for its removal
+// from storage.
+// TODO(pav-kv): for WAG, add the truncated state and applied index. See #152845.
+type destroyReplicaInfo struct {
+	id   roachpb.FullReplicaID
+	desc *roachpb.RangeDescriptor
+}
+
 // snapWriteBuilder contains the data needed to prepare the on-disk state for a
 // snapshot.
 type snapWriteBuilder struct {
@@ -29,10 +37,10 @@ type snapWriteBuilder struct {
 	sl       stateloader.StateLoader
 	writeSST func(context.Context, func(context.Context, storage.Writer) error) error
 
-	truncState    kvserverpb.RaftTruncatedState
-	hardState     raftpb.HardState
-	desc          *roachpb.RangeDescriptor
-	subsumedDescs []*roachpb.RangeDescriptor
+	truncState kvserverpb.RaftTruncatedState
+	hardState  raftpb.HardState
+	desc       *roachpb.RangeDescriptor
+	subsume    []destroyReplicaInfo
 
 	// cleared contains the spans that this snapshot application clears before
 	// writing new state on top.
@@ -115,7 +123,7 @@ func (s *snapWriteBuilder) clearSubsumedReplicaDiskData(ctx context.Context) err
 	}
 	keySpans := getKeySpans(s.desc)
 	totalKeySpans := append([]roachpb.Span(nil), keySpans...)
-	for _, subDesc := range s.subsumedDescs {
+	for _, sub := range s.subsume {
 		// We have to create an SST for the subsumed replica's range-id local keys.
 		if err := s.writeSST(ctx, func(ctx context.Context, w storage.Writer) error {
 			// NOTE: We set mustClearRange to true because we are setting
@@ -126,16 +134,16 @@ func (s *snapWriteBuilder) clearSubsumedReplicaDiskData(ctx context.Context) err
 				ClearUnreplicatedByRangeID: true,
 				MustUseClearRange:          true,
 			}
-			s.cleared = append(s.cleared, rditer.Select(subDesc.RangeID, rditer.SelectOpts{
+			s.cleared = append(s.cleared, rditer.Select(sub.id.RangeID, rditer.SelectOpts{
 				ReplicatedByRangeID:   opts.ClearReplicatedByRangeID,
 				UnreplicatedByRangeID: opts.ClearUnreplicatedByRangeID,
 			})...)
-			return kvstorage.DestroyReplica(ctx, subDesc.RangeID, reader, w, mergedTombstoneReplicaID, opts)
+			return kvstorage.DestroyReplica(ctx, sub.id, reader, w, mergedTombstoneReplicaID, opts)
 		}); err != nil {
 			return err
 		}
 
-		srKeySpans := getKeySpans(subDesc)
+		srKeySpans := getKeySpans(sub.desc)
 		// Compute the total key space covered by the current replica and all
 		// subsumed replicas.
 		for i := range srKeySpans {

pkg/kv/kvserver/snapshot_apply_prepare_test.go

@@ -69,7 +69,8 @@ func TestPrepareSnapApply(t *testing.T) {
 		}
 	}
 
-	id := roachpb.FullReplicaID{RangeID: 123, ReplicaID: 4}
+	const replicaID = 4
+	id := roachpb.FullReplicaID{RangeID: 123, ReplicaID: replicaID}
 	descA := desc(101, "a", "b")
 	descB := desc(102, "b", "z")
 	createRangeData(t, eng, *descA)
@@ -79,7 +80,7 @@ func TestPrepareSnapApply(t *testing.T) {
 	ctx := context.Background()
 	require.NoError(t, sl.SetRaftReplicaID(ctx, eng, id.ReplicaID))
 	for _, rID := range []roachpb.RangeID{101, 102} {
-		require.NoError(t, stateloader.Make(rID).SetRaftReplicaID(ctx, eng, id.ReplicaID))
+		require.NoError(t, stateloader.Make(rID).SetRaftReplicaID(ctx, eng, replicaID))
 	}
 
 	swb := snapWriteBuilder{
@@ -88,10 +89,13 @@ func TestPrepareSnapApply(t *testing.T) {
 		sl:       sl,
 		writeSST: writeSST,
 
-		truncState:    kvserverpb.RaftTruncatedState{Index: 100, Term: 20},
-		hardState:     raftpb.HardState{Term: 20, Commit: 100},
-		desc:          desc(id.RangeID, "a", "k"),
-		subsumedDescs: []*roachpb.RangeDescriptor{descA, descB},
+		truncState: kvserverpb.RaftTruncatedState{Index: 100, Term: 20},
+		hardState:  raftpb.HardState{Term: 20, Commit: 100},
+		desc:       desc(id.RangeID, "a", "k"),
+		subsume: []destroyReplicaInfo{
+			{id: roachpb.FullReplicaID{RangeID: descA.RangeID, ReplicaID: replicaID}, desc: descA},
+			{id: roachpb.FullReplicaID{RangeID: descB.RangeID, ReplicaID: replicaID}, desc: descB},
+		},
 	}
 
 	err := swb.prepareSnapApply(ctx)