storage: fix checksum version check

The check was sitting at evaluation time, where it is useless. It needs
to sit in the below-Raft code that actually computes the checksums.

This flew under the radar for quite some time, but was found in:

cockroachdb#37737 (comment)

thanks to the aggressive consistency checks we run in version/mixed/nodes=3.

Release note: None

pkg/storage/batcheval/cmd_compute_checksum.go

@@ -23,7 +23,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/storage/engine"
 	"github.com/cockroachdb/cockroach/pkg/storage/spanset"
 	"github.com/cockroachdb/cockroach/pkg/storage/storagepb"
-	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 )
 
@@ -54,17 +53,12 @@ func ComputeChecksum(
 ) (result.Result, error) {
 	args := cArgs.Args.(*roachpb.ComputeChecksumRequest)
 
-	if args.Version != ReplicaChecksumVersion {
-		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 = &storagepb.ComputeChecksum{
+		Version:      args.Version,
 		ChecksumID:   reply.ChecksumID,
 		SaveSnapshot: args.Snapshot,
 		Mode:         args.Mode,

pkg/storage/replica_consistency.go

@@ -401,12 +401,11 @@ func (r *Replica) getChecksum(ctx context.Context, id uuid.UUID) (ReplicaChecksu
 	r.mu.RLock()
 	c, ok = r.mu.checksums[id]
 	r.mu.RUnlock()
-	if !ok {
-		return ReplicaChecksum{}, errors.Errorf("no map entry for checksum (ID = %s)", id)
-	}
-	if c.Checksum == nil {
-		return ReplicaChecksum{}, errors.Errorf(
-			"checksum is nil, most likely because the async computation could not be run (ID = %s)", id)
+	// If the checksum wasn't found or the checksum could not be computed, error out.
+	// The latter case can occur when there's a version mismatch or, more generally,
+	// when the (async) checksum computation fails.
+	if !ok || c.Checksum == nil {
+		return ReplicaChecksum{}, errors.Errorf("no checksum found (ID = %s)", id)
 	}
 	return c, nil
 }

pkg/storage/replica_consistency_test.go

@@ -0,0 +1,62 @@
+// Copyright 2019 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package storage
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/storage/batcheval"
+	"github.com/cockroachdb/cockroach/pkg/storage/storagepb"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/stop"
+	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReplicaChecksumVersion(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx := context.TODO()
+	tc := testContext{}
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+	tc.Start(t, stopper)
+
+	testutils.RunTrueAndFalse(t, "matchingVersion", func(t *testing.T, matchingVersion bool) {
+		cc := storagepb.ComputeChecksum{
+			ChecksumID: uuid.FastMakeV4(),
+			Mode:       roachpb.ChecksumMode_CHECK_FULL,
+		}
+		if matchingVersion {
+			cc.Version = batcheval.ReplicaChecksumVersion
+		} else {
+			cc.Version = 1
+		}
+		tc.repl.computeChecksumPostApply(ctx, cc)
+		rc, err := tc.repl.getChecksum(ctx, cc.ChecksumID)
+		if !matchingVersion {
+			if !testutils.IsError(err, "no checksum found") {
+				t.Fatal(err)
+			}
+			require.Nil(t, rc.Checksum)
+		} else {
+			require.NoError(t, err)
+			require.NotNil(t, rc.Checksum)
+		}
+	})
+}

pkg/storage/replica_proposal.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/storage/batcheval"
 	"github.com/cockroachdb/cockroach/pkg/storage/batcheval/result"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
@@ -166,6 +167,14 @@ func (r *Replica) computeChecksumPostApply(ctx context.Context, cc storagepb.Com
 	r.mu.checksums[cc.ChecksumID] = ReplicaChecksum{started: true, notify: notify}
 	desc := *r.mu.state.Desc
 	r.mu.Unlock()
+
+	if cc.Version != batcheval.ReplicaChecksumVersion {
+		r.computeChecksumDone(ctx, cc.ChecksumID, nil, nil)
+		log.Infof(ctx, "incompatible ComputeChecksum versions (requested: %d, have: %d)",
+			cc.Version, batcheval.ReplicaChecksumVersion)
+		return
+	}
+
 	// Caller is holding raftMu, so an engine snapshot is automatically
 	// Raft-consistent (i.e. not in the middle of an AddSSTable).
 	snap := r.store.engine.NewSnapshot()

pkg/storage/replica_test.go

@@ -6847,32 +6847,6 @@ func TestReplicaTryAbandon(t *testing.T) {
 	})
 }
 
-// TestComputeChecksumVersioning checks that the ComputeChecksum post-commit
-// trigger is called if and only if the checksum version is right.
-func TestComputeChecksumVersioning(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	tc := testContext{}
-	stopper := stop.NewStopper()
-	defer stopper.Stop(context.TODO())
-	tc.Start(t, stopper)
-
-	if pct, _ := batcheval.ComputeChecksum(context.TODO(), nil,
-		batcheval.CommandArgs{Args: &roachpb.ComputeChecksumRequest{
-			Version: batcheval.ReplicaChecksumVersion,
-		}}, &roachpb.ComputeChecksumResponse{},
-	); pct.Replicated.ComputeChecksum == nil {
-		t.Error("right checksum version: expected post-commit trigger")
-	}
-
-	if pct, _ := batcheval.ComputeChecksum(context.TODO(), nil,
-		batcheval.CommandArgs{Args: &roachpb.ComputeChecksumRequest{
-			Version: batcheval.ReplicaChecksumVersion + 1,
-		}}, &roachpb.ComputeChecksumResponse{},
-	); pct.Replicated.ComputeChecksum != nil {
-		t.Errorf("wrong checksum version: expected no post-commit trigger: %s", pct.Replicated.ComputeChecksum)
-	}
-}
-
 func TestNewReplicaCorruptionError(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	for i, tc := range []struct {