Merge #29126

29126: backport-2.1: another round of merge bug fixes r=tschottdorf a=benesch

Backport:
  * 1/1 commits from "storage: fix raft snapshots that span merges and splits" (#29083)
  * 1/1 commits from "storage: deflake TestStoreRangeMergeReadoptedBothFollowers" (#29084)
  * 1/1 commits from "storage: protect ComputeChecksum commands from replaying" (#29067)
  * 1/1 commits from "storage: make ComputeChecksum a point request" (#29079)

Please see individual PRs for details.

/cc @cockroachdb/release


Co-authored-by: Nikhil Benesch <[email protected]>

Author: craig[bot]

pkg/internal/client/batch.go

@@ -516,24 +516,6 @@ func (b *Batch) ReverseScan(s, e interface{}) {
 	b.scan(s, e, true)
 }
 
-// CheckConsistency creates a batch request to check the consistency of the
-// ranges holding the span of keys from s to e. It logs a diff of all the
-// keys that are inconsistent when withDiff is set to true.
-func (b *Batch) CheckConsistency(s, e interface{}, withDiff bool) {
-	begin, err := marshalKey(s)
-	if err != nil {
-		b.initResult(0, 0, notRaw, err)
-		return
-	}
-	end, err := marshalKey(e)
-	if err != nil {
-		b.initResult(0, 0, notRaw, err)
-		return
-	}
-	b.appendReqs(roachpb.NewCheckConsistency(begin, end, withDiff))
-	b.initResult(1, 0, notRaw, nil)
-}
-
 // Del deletes one or more keys.
 //
 // A new result will be appended to the batch and each key will have a

pkg/internal/client/db.go

@@ -506,15 +506,6 @@ func (db *DB) AdminChangeReplicas(
 	return getOneErr(db.Run(ctx, b), b)
 }
 
-// CheckConsistency runs a consistency check on all the ranges containing
-// the key span. It logs a diff of all the keys that are inconsistent
-// when withDiff is set to true.
-func (db *DB) CheckConsistency(ctx context.Context, begin, end interface{}, withDiff bool) error {
-	b := &Batch{}
-	b.CheckConsistency(begin, end, withDiff)
-	return getOneErr(db.Run(ctx, b), b)
-}
-
 // WriteBatch applies the operations encoded in a BatchRepr, which is the
 // serialized form of a RocksDB Batch. The command cannot span Ranges and must
 // be run on an empty keyrange.

pkg/roachpb/api.go

@@ -930,17 +930,6 @@ func NewScan(key, endKey Key) Request {
 	}
 }
 
-// NewCheckConsistency returns a Request initialized to scan from start to end keys.
-func NewCheckConsistency(key, endKey Key, withDiff bool) Request {
-	return &CheckConsistencyRequest{
-		RequestHeader: RequestHeader{
-			Key:    key,
-			EndKey: endKey,
-		},
-		WithDiff: withDiff,
-	}
-}
-
 // NewReverseScan returns a Request initialized to reverse scan from end to
 // start keys with max results.
 func NewReverseScan(key, endKey Key) Request {
@@ -1064,7 +1053,7 @@ func (*TransferLeaseRequest) flags() int {
 	return isWrite | isAlone | skipLeaseCheck
 }
 func (*RecomputeStatsRequest) flags() int   { return isWrite | isAlone }
-func (*ComputeChecksumRequest) flags() int  { return isWrite | isRange }
+func (*ComputeChecksumRequest) flags() int  { return isWrite }
 func (*CheckConsistencyRequest) flags() int { return isAdmin | isRange }
 func (*WriteBatchRequest) flags() int       { return isWrite | isRange }
 func (*ExportRequest) flags() int           { return isRead | isRange | updatesReadTSCache }

pkg/roachpb/api.pb.go

@@ -991,11 +991,7 @@ message ComputeChecksumRequest {
   // The version used to pick the checksum method. It allows us to use a
   // consistent checksumming method across replicas.
   uint32 version = 2;
-  // A unique identifier to match a future storage.CollectChecksumRequest with
-  // this request.
-  bytes checksum_id = 3 [(gogoproto.nullable) = false,
-      (gogoproto.customname) = "ChecksumID",
-      (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID"];
+  reserved 3;
   // Compute a checksum along with a snapshot of the entire range, that will be
   // used in logging a diff during checksum verification.
   bool snapshot = 4;
@@ -1004,6 +1000,12 @@ message ComputeChecksumRequest {
 // A ComputeChecksumResponse is the response to a ComputeChecksum() operation.
 message ComputeChecksumResponse {
   ResponseHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+
+  // ChecksumID is the unique identifier that can be used to get the computed
+  // checksum in a future storage.CollectChecksumRequest.
+  bytes checksum_id = 2 [(gogoproto.nullable) = false,
+      (gogoproto.customname) = "ChecksumID",
+      (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID"];
 }
 
 enum ExportStorageProvider {

pkg/roachpb/api.proto

@@ -188,6 +188,16 @@ func (ba *BatchRequest) IsSingleSubsumeRequest() bool {
 	return false
 }
 
+// IsSingleComputeChecksumRequest returns true iff the batch contains a single
+// request, and that request is a ComputeChecksumRequest.
+func (ba *BatchRequest) IsSingleComputeChecksumRequest() bool {
+	if ba.IsSingleRequest() {
+		_, ok := ba.Requests[0].GetInner().(*ComputeChecksumRequest)
+		return ok
+	}
+	return false
+}
+
 // GetPrevLeaseForLeaseRequest returns the previous lease, at the time
 // of proposal, for a request lease or transfer lease request. If the
 // batch does not contain a single lease request, this method will panic.

pkg/roachpb/batch.go

@@ -21,17 +21,28 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/batcheval/result"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine"
+	"github.com/cockroachdb/cockroach/pkg/storage/spanset"
+	"github.com/cockroachdb/cockroach/pkg/storage/storagebase"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 )
 
 func init() {
-	RegisterCommand(roachpb.ComputeChecksum, DefaultDeclareKeys, ComputeChecksum)
+	RegisterCommand(roachpb.ComputeChecksum, declareKeysComputeChecksum, ComputeChecksum)
 }
 
-// Version numbers for Replica checksum computation. Requests fail unless the
-// versions are compatible.
+func declareKeysComputeChecksum(
+	roachpb.RangeDescriptor, roachpb.Header, roachpb.Request, *spanset.SpanSet,
+) {
+	// Intentionally declare no keys, as ComputeChecksum does not need to be
+	// serialized with any other commands. It simply needs to be committed into
+	// the Raft log.
+}
+
+// Version numbers for Replica checksum computation. Requests silently no-op
+// unless the versions are compatible.
 const (
-	ReplicaChecksumVersion    = 2
+	ReplicaChecksumVersion    = 3
 	ReplicaChecksumGCInterval = time.Hour
 )
 
@@ -44,10 +55,18 @@ func ComputeChecksum(
 	args := cArgs.Args.(*roachpb.ComputeChecksumRequest)
 
 	if args.Version != ReplicaChecksumVersion {
-		log.Errorf(ctx, "Incompatible versions: e=%d, v=%d", ReplicaChecksumVersion, args.Version)
+		log.Infof(ctx, "incompatible ComputeChecksum versions (server: %d, requested: %d)",
+			ReplicaChecksumVersion, args.Version)
 		return result.Result{}, nil
 	}
+
+	reply := resp.(*roachpb.ComputeChecksumResponse)
+	reply.ChecksumID = uuid.MakeV4()
+
 	var pd result.Result
-	pd.Replicated.ComputeChecksum = args
+	pd.Replicated.ComputeChecksum = &storagebase.ComputeChecksum{
+		ChecksumID:   reply.ChecksumID,
+		SaveSnapshot: args.Snapshot,
+	}
 	return pd, nil
 }

pkg/storage/batcheval/cmd_compute_checksum.go

@@ -1727,6 +1727,7 @@ func TestStoreRangeMergeReadoptedBothFollowers(t *testing.T) {
 	mtc.Start(t, 3)
 	defer mtc.Stop()
 	store0, store2 := mtc.Store(0), mtc.Store(2)
+	distSender := mtc.distSenders[0]
 
 	// Create two ranges on all nodes.
 	mtc.replicateRange(roachpb.RangeID(1), 1, 2)
@@ -1735,16 +1736,26 @@ func TestStoreRangeMergeReadoptedBothFollowers(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Wait for store2 to hear about the split.
-	var lhsRepl2, rhsRepl2 *storage.Replica
-	testutils.SucceedsSoon(t, func() error {
-		lhsRepl2, err = store2.GetReplica(lhsDesc.RangeID)
-		if err != nil {
-			return err
+	// Wait for all stores to have fully processed the split.
+	for _, key := range []roachpb.Key{roachpb.Key("a"), roachpb.Key("b")} {
+		if _, pErr := client.SendWrapped(ctx, distSender, incrementArgs(key, 1)); pErr != nil {
+			t.Fatal(pErr)
 		}
-		rhsRepl2, err = store2.GetReplica(rhsDesc.RangeID)
-		return err
-	})
+		mtc.waitForValues(key, []int64{1, 1, 1})
+	}
+
+	lhsRepl0, err := store0.GetReplica(lhsDesc.RangeID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	lhsRepl2, err := store2.GetReplica(lhsDesc.RangeID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rhsRepl2, err := store2.GetReplica(rhsDesc.RangeID)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	// Abandon the two ranges on store2, but do not GC them.
 	mtc.unreplicateRange(lhsDesc.RangeID, 2)
@@ -1757,13 +1768,6 @@ func TestStoreRangeMergeReadoptedBothFollowers(t *testing.T) {
 		t.Fatal(pErr)
 	}
 
-	// Attempt to re-add the merged range to store2. The operation should fail
-	// because store2's LHS and RHS replicas intersect the merged range.
-	lhsRepl0, err := store0.GetReplica(lhsDesc.RangeID)
-	if err != nil {
-		t.Fatal(err)
-	}
-
 	addLHSRepl2 := func() error {
 		for r := retry.StartWithCtx(ctx, retry.Options{}); r.Next(); {
 			err := lhsRepl0.ChangeReplicas(ctx, roachpb.ADD_REPLICA, roachpb.ReplicationTarget{
@@ -1778,6 +1782,8 @@ func TestStoreRangeMergeReadoptedBothFollowers(t *testing.T) {
 		return nil
 	}
 
+	// Attempt to re-add the merged range to store2. The operation should fail
+	// because store2's LHS and RHS replicas intersect the merged range.
 	err = addLHSRepl2()
 	if exp := "cannot apply snapshot: snapshot intersects existing range"; !testutils.IsError(err, exp) {
 		t.Fatalf("expected %q error, but got %v", exp, err)
@@ -1903,22 +1909,25 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 
 	// Create three fully-caught-up, adjacent ranges on all three stores.
 	mtc.replicateRange(roachpb.RangeID(1), 1, 2)
-	splitKeys := []roachpb.Key{roachpb.Key("a"), roachpb.Key("b"), roachpb.Key("c")}
-	for _, key := range splitKeys {
-		if _, pErr := client.SendWrapped(ctx, distSender, adminSplitArgs(key)); pErr != nil {
-			t.Fatal(pErr)
+	splitKeys := []roachpb.Key{roachpb.Key("a"), roachpb.Key("b"), roachpb.Key("c"), roachpb.Key("d")}
+	for i, key := range splitKeys {
+		if i != len(splitKeys)-1 {
+			// We'll split the last range off later.
+			if _, pErr := client.SendWrapped(ctx, distSender, adminSplitArgs(key)); pErr != nil {
+				t.Fatal(pErr)
+			}
 		}
 		if _, pErr := client.SendWrapped(ctx, distSender, incrementArgs(key, 1)); pErr != nil {
 			t.Fatal(pErr)
 		}
 		mtc.waitForValues(key, []int64{1, 1, 1})
 	}
 
-	lhsRepl0 := store0.LookupReplica(roachpb.RKey("a"))
+	aRepl0 := store0.LookupReplica(roachpb.RKey("a"))
 
 	// Start dropping all Raft traffic to the first range on store1.
 	mtc.transport.Listen(store2.Ident.StoreID, &unreliableRaftHandler{
-		rangeID:            lhsRepl0.RangeID,
+		rangeID:            aRepl0.RangeID,
 		RaftMessageHandler: store2,
 	})
 
@@ -1929,6 +1938,13 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 		}
 	}
 
+	// Split [a, /Max) into [a, d) and [d, /Max). This means the Raft snapshot
+	// will span both a merge and a split.
+	lastSplitKey := splitKeys[len(splitKeys)-1]
+	if _, pErr := client.SendWrapped(ctx, distSender, adminSplitArgs(lastSplitKey)); pErr != nil {
+		t.Fatal(pErr)
+	}
+
 	// Truncate the logs of the LHS.
 	{
 		repl := store0.LookupReplica(roachpb.RKey("a"))
@@ -1970,10 +1986,15 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 	// Verify that the sets of keys in store0 and store2 are identical.
 	storeKeys0 := getEngineKeySet(t, store0.Engine())
 	storeKeys2 := getEngineKeySet(t, store2.Engine())
+	dRepl0 := store0.LookupReplica(roachpb.RKey("d"))
 	ignoreKey := func(k string) bool {
-		// Unreplicated keys for the two remaining ranges are allowed to differ.
-		return strings.HasPrefix(k, string(keys.MakeRangeIDUnreplicatedPrefix(roachpb.RangeID(1)))) ||
-			strings.HasPrefix(k, string(keys.MakeRangeIDUnreplicatedPrefix(lhsRepl0.RangeID)))
+		// Unreplicated keys for the remaining ranges are allowed to differ.
+		for _, id := range []roachpb.RangeID{1, aRepl0.RangeID, dRepl0.RangeID} {
+			if strings.HasPrefix(k, string(keys.MakeRangeIDUnreplicatedPrefix(id))) {
+				return true
+			}
+		}
+		return false
 	}
 	for k := range storeKeys0 {
 		if ignoreKey(k) {

pkg/storage/client_merge_test.go

@@ -32,11 +32,14 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/storage/engine"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/storage/stateloader"
+	"github.com/cockroachdb/cockroach/pkg/storage/storagebase"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 )
 
 // TestConsistencyQueueRequiresLive verifies the queue will not
@@ -101,7 +104,75 @@ func TestCheckConsistencyMultiStore(t *testing.T) {
 	}, &checkArgs); err != nil {
 		t.Fatal(err)
 	}
+}
+
+// TestCheckConsistencyReplay verifies that two ComputeChecksum requests with
+// the same checksum ID are not committed to the Raft log, even if DistSender
+// retries the request.
+func TestCheckConsistencyReplay(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	type applyKey struct {
+		checksumID uuid.UUID
+		storeID    roachpb.StoreID
+	}
+	var state struct {
+		syncutil.Mutex
+		forcedRetry bool
+		applies     map[applyKey]int
+	}
+	state.applies = map[applyKey]int{}
+
+	var mtc *multiTestContext
+	ctx := context.Background()
+	storeCfg := storage.TestStoreConfig(nil /* clock */)
+
+	// Arrange to count the number of times each checksum command applies to each
+	// store.
+	storeCfg.TestingKnobs.TestingApplyFilter = func(args storagebase.ApplyFilterArgs) *roachpb.Error {
+		state.Lock()
+		defer state.Unlock()
+		if ccr := args.ComputeChecksum; ccr != nil {
+			state.applies[applyKey{ccr.ChecksumID, args.StoreID}]++
+		}
+		return nil
+	}
+
+	// Arrange to trigger a retry when a ComputeChecksum request arrives.
+	storeCfg.TestingKnobs.TestingResponseFilter = func(ba roachpb.BatchRequest, br *roachpb.BatchResponse) *roachpb.Error {
+		state.Lock()
+		defer state.Unlock()
+		if ba.IsSingleComputeChecksumRequest() && !state.forcedRetry {
+			state.forcedRetry = true
+			return roachpb.NewError(roachpb.NewSendError("injected failure"))
+		}
+		return nil
+	}
+
+	mtc = &multiTestContext{storeConfig: &storeCfg}
+	defer mtc.Stop()
+	mtc.Start(t, 2)
+
+	mtc.replicateRange(roachpb.RangeID(1), 1)
+
+	checkArgs := roachpb.CheckConsistencyRequest{
+		RequestHeader: roachpb.RequestHeader{
+			Key:    []byte("a"),
+			EndKey: []byte("b"),
+		},
+	}
+	if _, err := client.SendWrapped(ctx, mtc.Store(0).TestSender(), &checkArgs); err != nil {
+		t.Fatal(err)
+	}
 
+	state.Lock()
+	defer state.Unlock()
+	for applyKey, count := range state.applies {
+		if count != 1 {
+			t.Errorf("checksum %s was applied %d times to s%d (expected once)",
+				applyKey.checksumID, count, applyKey.storeID)
+		}
+	}
 }
 
 func TestCheckConsistencyInconsistent(t *testing.T) {