Merge #155436

155436: kvserver: add replica info struct for destruction r=arulajmani a=pav-kv

Related to #152845

Co-authored-by: Pavel Kalinnikov <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/kvstorage/destroy.go

@@ -108,49 +108,60 @@ func ClearRangeData(
 // atomically, and 1 is not written.
 const DestroyReplicaTODO = 0
 
+// DestroyReplicaInfo contains the replica's metadata needed for its removal
+// from storage.
+//
+// TODO(pav-kv): for separated storage, add the applied raft log span. #152845
+type DestroyReplicaInfo struct {
+	// FullReplicaID identifies the replica on its store.
+	roachpb.FullReplicaID
+	// Keys is the user key span of this replica, taken from its RangeDescriptor.
+	// Non-empty iff the replica is initialized.
+	Keys roachpb.RSpan
+}
+
 // DestroyReplica destroys all or a part of the Replica's state, installing a
-// RangeTombstone in its place. Due to merges, splits, etc, there is a need
-// to control which part of the state this method actually gets to remove,
-// which is done via the provided options[^1]; the caller is always responsible
-// for managing the remaining disk state accordingly.
+// RangeTombstone in its place. Due to merges, splits, etc, there is a need to
+// control which part of the state this method actually gets to remove, which is
+// done via the provided options[^1]; the caller is always responsible for
+// managing the remaining disk state accordingly.
 //
 // [^1] e.g., on a merge, the user data moves to the subsuming replica and must
 // not be cleared.
 func DestroyReplica(
 	ctx context.Context,
-	id roachpb.FullReplicaID,
 	reader storage.Reader,
 	writer storage.Writer,
-	nextReplicaID roachpb.ReplicaID,
+	info DestroyReplicaInfo,
+	next roachpb.ReplicaID,
 	opts ClearRangeDataOptions,
 ) error {
-	diskReplicaID, err := stateloader.Make(id.RangeID).LoadRaftReplicaID(ctx, reader)
-	if err != nil {
-		return err
+	if next <= info.ReplicaID {
+		return errors.AssertionFailedf("%v must not survive its own tombstone", info.FullReplicaID)
 	}
-	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)
+	sl := stateloader.Make(info.RangeID)
+	// Assert that the ReplicaID in storage matches the one being removed.
+	if loaded, err := sl.LoadRaftReplicaID(ctx, reader); err != nil {
+		return err
+	} else if id := loaded.ReplicaID; id != info.ReplicaID {
+		return errors.AssertionFailedf("%v has a mismatching ID %d", info.FullReplicaID, id)
 	}
-	_ = DestroyReplicaTODO // 2.1 + 2.2 + 3.1
-	if err := ClearRangeData(ctx, id.RangeID, reader, writer, opts); err != nil {
+	// Assert that the provided tombstone moves the existing one strictly forward.
+	// A failure would indicate that something is wrong in the replica lifecycle.
+	if ts, err := sl.LoadRangeTombstone(ctx, reader); err != nil {
 		return err
+	} else if next <= ts.NextReplicaID {
+		return errors.AssertionFailedf("%v cannot rewind tombstone from %d to %d",
+			info.FullReplicaID, ts.NextReplicaID, next)
 	}
 
-	// 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(id.RangeID)
-	ts, err := sl.LoadRangeTombstone(ctx, reader)
-	if err != nil {
+	_ = DestroyReplicaTODO // 2.1 + 2.2 + 3.1
+	if err := ClearRangeData(ctx, info.RangeID, reader, writer, opts); err != nil {
 		return err
-	} else if ts.NextReplicaID >= nextReplicaID {
-		return errors.AssertionFailedf(
-			"cannot rewind tombstone from %d to %d", ts.NextReplicaID, nextReplicaID)
 	}
+	// Save a tombstone to ensure that replica IDs never get reused.
 	_ = DestroyReplicaTODO // 2.3
 	return sl.SetRangeTombstone(ctx, writer, kvserverpb.RangeTombstone{
-		NextReplicaID: nextReplicaID, // NB: nextReplicaID > 0
+		NextReplicaID: next, // NB: NextReplicaID > 0
 	})
 }

pkg/kv/kvserver/kvstorage/destroy_test.go

@@ -68,11 +68,15 @@ func TestDestroyReplica(t *testing.T) {
 		r.createStateMachine(ctx, t, rw)
 	})
 	mutate("destroy", func(rw storage.ReadWriter) {
-		require.NoError(t, DestroyReplica(ctx, r.id, rw, rw, r.id.ReplicaID+1, ClearRangeDataOptions{
-			ClearUnreplicatedByRangeID: true,
-			ClearReplicatedByRangeID:   true,
-			ClearReplicatedBySpan:      r.keys,
-		}))
+		require.NoError(t, DestroyReplica(
+			ctx, rw, rw,
+			DestroyReplicaInfo{FullReplicaID: r.id, Keys: r.keys}, r.id.ReplicaID+1,
+			ClearRangeDataOptions{
+				ClearUnreplicatedByRangeID: true,
+				ClearReplicatedByRangeID:   true,
+				ClearReplicatedBySpan:      r.keys,
+			},
+		))
 	})
 
 	str := strings.ReplaceAll(sb.String(), "\n\n", "\n")

pkg/kv/kvserver/replica_app_batch.go

@@ -364,10 +364,15 @@ 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.ID(), b.batch, b.batch, mergedTombstoneReplicaID, kvstorage.ClearRangeDataOptions{
-			ClearReplicatedByRangeID:   true,
-			ClearUnreplicatedByRangeID: true,
-		}); err != nil {
+		if err := kvstorage.DestroyReplica(
+			ctx, b.batch, b.batch,
+			rhsRepl.destroyInfoRaftMuLocked(),
+			mergedTombstoneReplicaID,
+			kvstorage.ClearRangeDataOptions{
+				ClearReplicatedByRangeID:   true,
+				ClearUnreplicatedByRangeID: true,
+			},
+		); err != nil {
 			return errors.Wrapf(err, "unable to destroy replica before merge")
 		}
 
@@ -455,11 +460,16 @@ 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.ID(), b.batch, b.batch, change.NextReplicaID(), kvstorage.ClearRangeDataOptions{
-			ClearReplicatedBySpan:      span,
-			ClearReplicatedByRangeID:   true,
-			ClearUnreplicatedByRangeID: true,
-		}); err != nil {
+		if err := kvstorage.DestroyReplica(
+			ctx, b.batch, b.batch,
+			b.r.destroyInfoRaftMuLocked(),
+			change.NextReplicaID(),
+			kvstorage.ClearRangeDataOptions{
+				ClearReplicatedBySpan:      span,
+				ClearReplicatedByRangeID:   true,
+				ClearUnreplicatedByRangeID: true,
+			},
+		); err != nil {
 			return errors.Wrapf(err, "unable to destroy replica before removal")
 		}
 	}

pkg/kv/kvserver/replica_destroy.go

@@ -111,7 +111,10 @@ 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.ID(), r.store.TODOEngine(), batch, nextReplicaID, opts); err != nil {
+	if err := kvstorage.DestroyReplica(
+		ctx, r.store.TODOEngine(), batch,
+		r.destroyInfoRaftMuLocked(), nextReplicaID, opts,
+	); err != nil {
 		return err
 	}
 	preTime := timeutil.Now()

pkg/kv/kvserver/replica_raftstorage.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage/snaprecv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/rditer"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary"
@@ -568,7 +569,7 @@ func (r *Replica) applySnapshotRaftMuLocked(
 		Term:  kvpb.RaftTerm(nonemptySnap.Metadata.Term),
 	}
 
-	subsume := make([]destroyReplicaInfo, 0, len(subsumedRepls))
+	subsume := make([]kvstorage.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
@@ -582,19 +583,18 @@ 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()
 
-		subsume = append(subsume, destroyReplicaInfo{id: sr.ID(), desc: srDesc})
+		subsume = append(subsume, sr.destroyInfoRaftMuLocked())
 	}
 
-	// NB: subsumedDescs in snapWriteBuilder must be sorted by start key. This
+	// NB: subsumed replicas in snapWriteBuilder must be sorted by start key. This
 	// should be the case, by construction, but add a test-only assertion just in
 	// case this ever changes.
-	testingAssert(slices.IsSortedFunc(subsume, func(a, b destroyReplicaInfo) int {
-		return a.desc.StartKey.Compare(b.desc.StartKey)
-	}), "subsumedDescs must be sorted by start key")
+	testingAssert(slices.IsSortedFunc(subsume, func(a, b kvstorage.DestroyReplicaInfo) int {
+		return a.Keys.Key.Compare(b.Keys.Key)
+	}), "subsumed replicas must be sorted by start key")
 
 	sb := snapWriteBuilder{
 		id: r.ID(),
@@ -842,3 +842,17 @@ func testingAssert(cond bool, msg string) {
 		panic(msg)
 	}
 }
+
+// destroyInfoRaftMuLocked returns the information necessary for constructing a
+// storage write destroying this replica.
+//
+// NB: since raftMu is locked, there is no concurrent write that would be able
+// to change this replica. In particular, no concurrent log truncations. The
+// caller must make sure to complete the destruction before raftMu is released.
+func (r *Replica) destroyInfoRaftMuLocked() kvstorage.DestroyReplicaInfo {
+	r.raftMu.AssertHeld()
+	return kvstorage.DestroyReplicaInfo{
+		FullReplicaID: r.ID(),
+		Keys:          r.shMu.state.Desc.RSpan(),
+	}
+}

pkg/kv/kvserver/snapshot_apply_prepare.go

@@ -20,14 +20,6 @@ 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 {
@@ -41,9 +33,8 @@ type snapWriteBuilder struct {
 	hardState  raftpb.HardState
 	desc       *roachpb.RangeDescriptor // corresponds to the range descriptor in the snapshot
 	origDesc   *roachpb.RangeDescriptor // pre-snapshot range descriptor
-	// NB: subsume, if set, must be in sorted (by destroyReplicaInfo.desc start
-	// key) order.
-	subsume []destroyReplicaInfo
+	// NB: subsume must be in sorted order by DestroyReplicaInfo start key.
+	subsume []kvstorage.DestroyReplicaInfo
 
 	// cleared contains the spans that this snapshot application clears before
 	// writing new state on top.
@@ -112,7 +103,7 @@ func (s *snapWriteBuilder) rewriteRaftState(ctx context.Context, w storage.Write
 // the Reader reflects the latest I/O each of the subsumed replicas has done
 // (i.e. Reader was instantiated after all raftMu were acquired).
 //
-// NB: does nothing if s.subsumedDescs is empty.
+// NB: does nothing if there are no subsumed replicas.
 func (s *snapWriteBuilder) clearSubsumedReplicaDiskData(ctx context.Context) error {
 	if len(s.subsume) == 0 {
 		return nil // no subsumed replicas to speak of; early return
@@ -123,10 +114,10 @@ func (s *snapWriteBuilder) clearSubsumedReplicaDiskData(ctx context.Context) err
 	// the left implies that either we merged "to the left" (we don't), or that
 	// we're applying a snapshot for another range (we don't do that either).
 	// Something is severely wrong for this to happen, so perform a sanity check.
-	if s.subsume[0].desc.StartKey.Compare(s.desc.StartKey) < 0 { // subsumedDescs are sorted by StartKey
+	if s.subsume[0].Keys.Key.Compare(s.desc.StartKey) < 0 { // subsume is sorted by start key
 		log.KvDistribution.Fatalf(ctx,
 			"subsuming replica to our left; subsumed desc start key: %v; snapshot desc start key %v",
-			s.subsume[0].desc.StartKey, s.desc.StartKey,
+			s.subsume[0].Keys.Key, s.desc.StartKey,
 		)
 	}
 
@@ -167,12 +158,12 @@ func (s *snapWriteBuilder) clearSubsumedReplicaDiskData(ctx context.Context) err
 				ClearUnreplicatedByRangeID: true,
 				MustUseClearRange:          true,
 			}
-			s.cleared = append(s.cleared, rditer.Select(sub.id.RangeID, rditer.SelectOpts{
+			s.cleared = append(s.cleared, rditer.Select(sub.RangeID, rditer.SelectOpts{
 				ReplicatedByRangeID:   opts.ClearReplicatedByRangeID,
 				UnreplicatedByRangeID: opts.ClearUnreplicatedByRangeID,
 			})...)
 			// NB: Actually clear RangeID local key spans.
-			return kvstorage.DestroyReplica(ctx, sub.id, reader, w, mergedTombstoneReplicaID, opts)
+			return kvstorage.DestroyReplica(ctx, reader, w, sub, mergedTombstoneReplicaID, opts)
 		}); err != nil {
 			return err
 		}
@@ -223,23 +214,22 @@ func (s *snapWriteBuilder) clearResidualDataOnNarrowSnapshot(ctx context.Context
 		return nil
 	}
 
-	rightMostDesc := s.origDesc
+	endKey := s.origDesc.EndKey
 	if len(s.subsume) != 0 {
 		// NB: s.subsume are non-overlapping and sorted by start key. Pick the last
 		// one to determine whether the snapshot is narrowing the keyspace or not.
-		rightMostDesc = s.subsume[len(s.subsume)-1].desc
+		endKey = s.subsume[len(s.subsume)-1].Keys.EndKey
 	}
 
-	if rightMostDesc.EndKey.Compare(s.desc.EndKey) <= 0 {
+	if endKey.Compare(s.desc.EndKey) <= 0 {
 		return nil // we aren't narrowing anything; no-op
 	}
 
 	// TODO(sep-raft-log): read from the state machine engine here.
 	reader := storage.Reader(s.todoEng)
 	for _, span := range rditer.Select(0, rditer.SelectOpts{
-		Ranged: rditer.SelectRangedOptions{RSpan: roachpb.RSpan{
-			Key: s.desc.EndKey, EndKey: rightMostDesc.EndKey,
-		},
+		Ranged: rditer.SelectRangedOptions{
+			RSpan:      roachpb.RSpan{Key: s.desc.EndKey, EndKey: endKey},
 			SystemKeys: true,
 			LockTable:  true,
 			UserKeys:   true,

pkg/kv/kvserver/snapshot_apply_prepare_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/print"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/rditer"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/stateloader"
@@ -93,9 +94,9 @@ func TestPrepareSnapApply(t *testing.T) {
 		hardState:  raftpb.HardState{Term: 20, Commit: 100},
 		desc:       desc(id.RangeID, "a", "k"),
 		origDesc:   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},
+		subsume: []kvstorage.DestroyReplicaInfo{
+			{FullReplicaID: roachpb.FullReplicaID{RangeID: descA.RangeID, ReplicaID: replicaID}, Keys: descA.RSpan()},
+			{FullReplicaID: roachpb.FullReplicaID{RangeID: descB.RangeID, ReplicaID: replicaID}, Keys: descB.RSpan()},
 		},
 	}