kvnemesis: add Raft application assertions

This adds basic kvnemesis assertions for Raft command application:

* A request is only applied once, at a single log index and lease
  applied index across all replica (i.e. no double-applies or replays).

* A replica's applied index, applied term, lease applied index, and
  closed timestamp do not regress (assuming no node restarts).

Epic: none
Release note: None

pkg/kv/kvnemesis/BUILD.bazel

@@ -79,7 +79,10 @@ go_test(
         "//pkg/kv/kvnemesis/kvnemesisutil",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
+        "//pkg/kv/kvserver/apply",
         "//pkg/kv/kvserver/concurrency/isolation",
+        "//pkg/kv/kvserver/kvserverbase",
+        "//pkg/kv/kvserver/kvserverpb",
         "//pkg/roachpb",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",

pkg/kv/kvnemesis/kvnemesis_test.go

@@ -23,6 +23,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/apply"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
@@ -131,6 +134,28 @@ func (cfg kvnemesisTestCfg) testClusterArgs(tr *SeqTracker) base.TestClusterArgs
 		}
 	}
 
+	if cfg.assertRaftApply {
+		asserter := apply.NewAsserter()
+		storeKnobs.TestingProposalSubmitFilter = func(args kvserverbase.ProposalFilterArgs) (bool, error) {
+			asserter.Propose(args.RangeID, args.ReplicaID, args.CmdID, args.SeedID, args.Cmd, args.Req)
+			return false /* drop */, nil
+		}
+		storeKnobs.TestingApplyCalledTwiceFilter = func(args kvserverbase.ApplyFilterArgs) (int, *kvpb.Error) {
+			if !args.Ephemeral {
+				asserter.Apply(args.RangeID, args.ReplicaID, args.CmdID, args.Entry, args.Cmd.MaxLeaseIndex,
+					*args.Cmd.ClosedTimestamp)
+			}
+			return 0, nil
+		}
+		storeKnobs.AfterSnapshotApplication = func(
+			desc roachpb.ReplicaDescriptor, state kvserverpb.ReplicaState, snap kvserver.IncomingSnapshot,
+		) {
+			asserter.ApplySnapshot(snap.Desc.RangeID, desc.ReplicaID, snap.FromReplica.ReplicaID,
+				state.RaftAppliedIndex, state.RaftAppliedIndexTerm, state.LeaseAppliedIndex,
+				state.RaftClosedTimestamp)
+		}
+	}
+
 	return base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Knobs: base.TestingKnobs{
@@ -228,6 +253,10 @@ type kvnemesisTestCfg struct {
 	// considered truly random, but is random enough for the desired purpose.
 	invalidLeaseAppliedIndexProb float64 // [0,1)
 	injectReproposalErrorProb    float64 // [0,1)
+	// If enabled, track Raft proposals and command application, and assert
+	// invariants (in particular that we don't double-apply a request or
+	// proposal).
+	assertRaftApply bool
 }
 
 func TestKVNemesisSingleNode(t *testing.T) {
@@ -241,6 +270,7 @@ func TestKVNemesisSingleNode(t *testing.T) {
 		seedOverride:                 0,
 		invalidLeaseAppliedIndexProb: 0.2,
 		injectReproposalErrorProb:    0.2,
+		assertRaftApply:              true,
 	})
 }
 
@@ -255,6 +285,7 @@ func TestKVNemesisSingleNode_ReproposalChaos(t *testing.T) {
 		seedOverride:                 0,
 		invalidLeaseAppliedIndexProb: 0.9,
 		injectReproposalErrorProb:    0.5,
+		assertRaftApply:              true,
 	})
 }
 
@@ -269,6 +300,7 @@ func TestKVNemesisMultiNode(t *testing.T) {
 		seedOverride:                 0,
 		invalidLeaseAppliedIndexProb: 0.2,
 		injectReproposalErrorProb:    0.2,
+		assertRaftApply:              true,
 	})
 }
 

pkg/kv/kvserver/apply/BUILD.bazel

@@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "apply",
     srcs = [
+        "asserter.go",
         "cmd.go",
         "doc.go",
         "task.go",
@@ -11,6 +12,11 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/kv/kvpb",
+        "//pkg/kv/kvserver/kvserverbase",
+        "//pkg/kv/kvserver/kvserverpb",
+        "//pkg/roachpb",
+        "//pkg/util/hlc",
+        "//pkg/util/syncutil",
         "@com_github_cockroachdb_errors//:errors",
         "@io_etcd_go_raft_v3//raftpb",
     ],