kv: Use strict types for common fields

This PR introduces 3 new typed fields to the kvpb package: RaftTerm,
RaftIndex and LeaseAppliedIndex. These fields were previously just
unit64 throughout the code and this made the code harder to read and
risked incorrect conversions.

It moves internal_raft.proto into the kvserver package to make this
possible to stay in the kv package.

Epic: none

Release note: None

pkg/cli/BUILD.bazel

@@ -128,6 +128,7 @@ go_library(
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/gc",
+        "//pkg/kv/kvserver/kvserverpb",
         "//pkg/kv/kvserver/kvstorage",
         "//pkg/kv/kvserver/liveness/livenesspb",
         "//pkg/kv/kvserver/loqrecovery",

pkg/cli/debug_check_store.go

@@ -19,6 +19,8 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/cli/clierrorplus"
 	"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/rditer"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/stateloader"
@@ -80,11 +82,11 @@ func runDebugCheckStoreCmd(cmd *cobra.Command, args []string) error {
 }
 
 type replicaCheckInfo struct {
-	truncatedIndex uint64
-	appliedIndex   uint64
-	firstIndex     uint64
-	lastIndex      uint64
-	committedIndex uint64
+	truncatedIndex kvpb.RaftIndex
+	appliedIndex   kvpb.RaftIndex
+	firstIndex     kvpb.RaftIndex
+	lastIndex      kvpb.RaftIndex
+	committedIndex kvpb.RaftIndex
 }
 
 type checkInput struct {
@@ -251,21 +253,22 @@ func checkStoreRaftState(
 				if err := kv.Value.GetProto(&hs); err != nil {
 					return err
 				}
-				getReplicaInfo(rangeID).committedIndex = hs.Commit
+				getReplicaInfo(rangeID).committedIndex = kvpb.RaftIndex(hs.Commit)
 			case bytes.Equal(suffix, keys.LocalRaftTruncatedStateSuffix):
-				var trunc roachpb.RaftTruncatedState
+				var trunc kvserverpb.RaftTruncatedState
 				if err := kv.Value.GetProto(&trunc); err != nil {
 					return err
 				}
 				getReplicaInfo(rangeID).truncatedIndex = trunc.Index
 			case bytes.Equal(suffix, keys.LocalRangeAppliedStateSuffix):
-				var state enginepb.RangeAppliedState
+				var state kvserverpb.RangeAppliedState
 				if err := kv.Value.GetProto(&state); err != nil {
 					return err
 				}
 				getReplicaInfo(rangeID).appliedIndex = state.RaftAppliedIndex
 			case bytes.Equal(suffix, keys.LocalRaftLogSuffix):
-				_, index, err := encoding.DecodeUint64Ascending(detail)
+				_, uIndex, err := encoding.DecodeUint64Ascending(detail)
+				index := kvpb.RaftIndex(uIndex)
 				if err != nil {
 					return err
 				}

pkg/keys/keys.go

@@ -356,21 +356,21 @@ func RaftLogPrefix(rangeID roachpb.RangeID) roachpb.Key {
 }
 
 // RaftLogKey returns a system-local key for a Raft log entry.
-func RaftLogKey(rangeID roachpb.RangeID, logIndex uint64) roachpb.Key {
+func RaftLogKey(rangeID roachpb.RangeID, logIndex kvpb.RaftIndex) roachpb.Key {
 	return MakeRangeIDPrefixBuf(rangeID).RaftLogKey(logIndex)
 }
 
 // RaftLogKeyFromPrefix returns a system-local key for a Raft log entry, using
 // the provided Raft log prefix.
-func RaftLogKeyFromPrefix(raftLogPrefix []byte, logIndex uint64) roachpb.Key {
-	return encoding.EncodeUint64Ascending(raftLogPrefix, logIndex)
+func RaftLogKeyFromPrefix(raftLogPrefix []byte, logIndex kvpb.RaftIndex) roachpb.Key {
+	return encoding.EncodeUint64Ascending(raftLogPrefix, uint64(logIndex))
 }
 
 // DecodeRaftLogKeyFromSuffix parses the suffix of a system-local key for a Raft
 // log entry and returns the entry's log index.
-func DecodeRaftLogKeyFromSuffix(raftLogSuffix []byte) (uint64, error) {
+func DecodeRaftLogKeyFromSuffix(raftLogSuffix []byte) (kvpb.RaftIndex, error) {
 	_, logIndex, err := encoding.DecodeUint64Ascending(raftLogSuffix)
-	return logIndex, err
+	return kvpb.RaftIndex(logIndex), err
 }
 
 // RaftReplicaIDKey returns a system-local key for a RaftReplicaID.
@@ -1092,7 +1092,7 @@ func (b RangeIDPrefixBuf) RaftLogPrefix() roachpb.Key {
 }
 
 // RaftLogKey returns a system-local key for a Raft log entry.
-func (b RangeIDPrefixBuf) RaftLogKey(logIndex uint64) roachpb.Key {
+func (b RangeIDPrefixBuf) RaftLogKey(logIndex kvpb.RaftIndex) roachpb.Key {
 	return RaftLogKeyFromPrefix(b.RaftLogPrefix(), logIndex)
 }
 

pkg/keys/printer.go

@@ -17,6 +17,7 @@ import (
 	"strings"
 	"unicode/utf8"
 
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
@@ -305,7 +306,7 @@ func raftLogKeyParse(rangeID roachpb.RangeID, input string) (string, roachpb.Key
 	if err != nil {
 		panic(err)
 	}
-	return "", RaftLogKey(rangeID, index)
+	return "", RaftLogKey(rangeID, kvpb.RaftIndex(index))
 }
 
 func raftLogKeyPrint(buf *redact.StringBuilder, key roachpb.Key) {

pkg/keys/printer_test.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/cockroachdb/apd/v3"
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/bitarray"
 	"github.com/cockroachdb/cockroach/pkg/util/duration"
@@ -250,7 +251,7 @@ func TestPrettyPrint(t *testing.T) {
 
 		{keys.RaftHardStateKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RaftHardState", revertSupportUnknown},
 		{keys.RangeTombstoneKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RangeTombstone", revertSupportUnknown},
-		{keys.RaftLogKey(roachpb.RangeID(1000001), uint64(200001)), "/Local/RangeID/1000001/u/RaftLog/logIndex:200001", revertSupportUnknown},
+		{keys.RaftLogKey(roachpb.RangeID(1000001), kvpb.RaftIndex(200001)), "/Local/RangeID/1000001/u/RaftLog/logIndex:200001", revertSupportUnknown},
 		{keys.RangeLastReplicaGCTimestampKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RangeLastReplicaGCTimestamp", revertSupportUnknown},
 
 		{keys.MakeRangeKeyPrefix(roachpb.RKey(tenSysCodec.TablePrefix(42))), `/Local/Range/Table/42`, revertSupportUnknown},

pkg/kv/kvpb/api.proto

@@ -1476,7 +1476,7 @@ message TruncateLogRequest {
   RequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
 
   // Log entries < this index are to be discarded.
-  uint64 index = 2;
+  uint64 index = 2  [(gogoproto.casttype) = "RaftIndex"];
 
   // RangeID is used to double check that the correct range is being truncated.
   // The header specifies a span, start and end keys, but not the range id
@@ -1496,7 +1496,7 @@ message TruncateLogRequest {
   // discarded from the raft log to be performed once, at the leaseholder.
   //
   // Populated starting at cluster version LooselyCoupledRaftLogTruncation.
-  uint64 expected_first_index = 4;
+  uint64 expected_first_index = 4  [(gogoproto.casttype) = "RaftIndex"];
 }
 
 // TruncateLogResponse is the response to a TruncateLog() operation.
@@ -2113,7 +2113,7 @@ message SubsumeResponse {
   // provide the sender of the Subsume request with an upper bound on the lease
   // applied index of the CPut that left an intent on the local copy of the
   // right-hand range descriptor.
-  uint64 lease_applied_index = 4;
+  uint64 lease_applied_index = 4  [(gogoproto.casttype) = "LeaseAppliedIndex"];
 
   // FreezeStart is a timestamp that is guaranteed to be greater than the
   // timestamps at which any requests were serviced by the responding replica

pkg/kv/kvpb/data.go

@@ -151,3 +151,25 @@ func TransactionRefreshTimestamp(pErr *Error) (bool, hlc.Timestamp) {
 	}
 	return true, timestamp
 }
+
+// LeaseAppliedIndex is attached to every Raft message and is used for replay
+// protection.
+type LeaseAppliedIndex uint64
+
+func (s LeaseAppliedIndex) SafeValue() {}
+
+// RaftTerm represents the term of a raft message. This corresponds to Term in
+// HardState.Term in the Raft library. That type is a uint64, so it is necessary
+// to cast to/from that type when dealing with the Raft library, however
+// internally RaftTerm is used for all fields in CRDB.
+type RaftTerm uint64
+
+// RaftIndex represents the term of a raft message. This corresponds to Index in
+// HardState.Index in the Raft library. That type is a uint64, so it is
+// necessary to cast to/from that type when dealing with the Raft library,
+// however internally RaftIndex is used for all fields in CRDB.
+type RaftIndex uint64
+
+// SafeValue implements the redact.SafeValue interface.
+func (s RaftTerm) SafeValue()  {}
+func (s RaftIndex) SafeValue() {}

pkg/kv/kvserver/BUILD.bazel

@@ -125,7 +125,6 @@ go_library(
         "//pkg/kv/kvserver/batcheval",
         "//pkg/kv/kvserver/batcheval/result",
         "//pkg/kv/kvserver/closedts",
-        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/kv/kvserver/closedts/sidetransport",
         "//pkg/kv/kvserver/closedts/tracker",
         "//pkg/kv/kvserver/concurrency",
@@ -364,7 +363,6 @@ go_test(
         "//pkg/kv/kvserver/batcheval",
         "//pkg/kv/kvserver/batcheval/result",
         "//pkg/kv/kvserver/closedts",
-        "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/kv/kvserver/closedts/tracker",
         "//pkg/kv/kvserver/concurrency",
         "//pkg/kv/kvserver/concurrency/isolation",

pkg/kv/kvserver/allocator/allocatorimpl/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/gossip",
+        "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/allocator",
         "//pkg/kv/kvserver/allocator/load",
         "//pkg/kv/kvserver/allocator/storepool",
@@ -46,6 +47,7 @@ go_test(
         "//pkg/config/zonepb",
         "//pkg/gossip",
         "//pkg/keys",
+        "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/allocator",
         "//pkg/kv/kvserver/allocator/load",
         "//pkg/kv/kvserver/allocator/storepool",

pkg/kv/kvserver/allocator/allocatorimpl/allocator.go

@@ -19,6 +19,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/load"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/storepool"
@@ -1957,7 +1958,7 @@ func (a *Allocator) ValidLeaseTargets(
 	leaseRepl interface {
 		StoreID() roachpb.StoreID
 		RaftStatus() *raft.Status
-		GetFirstIndex() uint64
+		GetFirstIndex() kvpb.RaftIndex
 		Desc() *roachpb.RangeDescriptor
 	},
 	opts allocator.TransferLeaseOptions,
@@ -2132,7 +2133,7 @@ func (a *Allocator) leaseholderShouldMoveDueToPreferences(
 	leaseRepl interface {
 		StoreID() roachpb.StoreID
 		RaftStatus() *raft.Status
-		GetFirstIndex() uint64
+		GetFirstIndex() kvpb.RaftIndex
 	},
 	allExistingReplicas []roachpb.ReplicaDescriptor,
 ) bool {
@@ -2217,7 +2218,7 @@ func (a *Allocator) TransferLeaseTarget(
 		StoreID() roachpb.StoreID
 		GetRangeID() roachpb.RangeID
 		RaftStatus() *raft.Status
-		GetFirstIndex() uint64
+		GetFirstIndex() kvpb.RaftIndex
 		Desc() *roachpb.RangeDescriptor
 	},
 	usageInfo allocator.RangeUsageInfo,
@@ -2485,7 +2486,7 @@ func (a *Allocator) ShouldTransferLease(
 	leaseRepl interface {
 		StoreID() roachpb.StoreID
 		RaftStatus() *raft.Status
-		GetFirstIndex() uint64
+		GetFirstIndex() kvpb.RaftIndex
 		Desc() *roachpb.RangeDescriptor
 	},
 	usageInfo allocator.RangeUsageInfo,
@@ -2852,7 +2853,10 @@ func FilterBehindReplicas(
 // Other replicas may be filtered out if this function is called with the
 // `raftStatus` of a non-raft leader replica.
 func excludeReplicasInNeedOfSnapshots(
-	ctx context.Context, st *raft.Status, firstIndex uint64, replicas []roachpb.ReplicaDescriptor,
+	ctx context.Context,
+	st *raft.Status,
+	firstIndex kvpb.RaftIndex,
+	replicas []roachpb.ReplicaDescriptor,
 ) []roachpb.ReplicaDescriptor {
 	filled := 0
 	for _, repl := range replicas {

pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
 	"github.com/cockroachdb/cockroach/pkg/gossip"
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/load"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/storepool"
@@ -1893,7 +1894,7 @@ func (r *mockRepl) RaftStatus() *raft.Status {
 	return raftStatus
 }
 
-func (r *mockRepl) GetFirstIndex() uint64 {
+func (r *mockRepl) GetFirstIndex() kvpb.RaftIndex {
 	return 0
 }
 

pkg/kv/kvserver/api.proto

@@ -55,7 +55,7 @@ message WaitForApplicationRequest {
   StoreRequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
   int64 range_id = 2 [(gogoproto.customname) = "RangeID",
       (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/roachpb.RangeID"];
-  uint64 lease_index = 3;
+  uint64 lease_index = 3  [(gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/kv/kvpb.LeaseAppliedIndex"];
 }
 
 message WaitForApplicationResponse {

pkg/kv/kvserver/app_batch.go

@@ -13,6 +13,7 @@ package kvserver
 import (
 	"context"
 
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/logstore"
@@ -179,7 +180,7 @@ func (b *appBatch) runPostAddTriggers(
 			env.st,
 			env.eng,
 			env.sideloaded,
-			cmd.Term,
+			kvpb.RaftTerm(cmd.Term),
 			cmd.Index(),
 			*res.AddSSTable,
 			env.bulkLimiter,

pkg/kv/kvserver/apply/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/apply",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/kv/kvpb",
         "@com_github_cockroachdb_errors//:errors",
         "@io_etcd_go_raft_v3//raftpb",
     ],
@@ -26,6 +27,7 @@ go_test(
     args = ["-test.timeout=55s"],
     deps = [
         ":apply",
+        "//pkg/kv/kvpb",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
         "@io_etcd_go_raft_v3//raftpb",

pkg/kv/kvserver/apply/cmd.go

@@ -10,14 +10,18 @@
 
 package apply
 
-import "context"
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+)
 
 // Command is a command that has been successfully replicated through raft
 // by being durably committed to the raft log of a quorum of peers in a raft
 // group.
 type Command interface {
 	// Index is the log index of the corresponding raft entry.
-	Index() uint64
+	Index() kvpb.RaftIndex
 	// IsTrivial returns whether the command can apply in a batch with other
 	// "trivial" commands. This is the case if the log entry represented by the
 	// Command is a simple write, as is the case for all user-issued mutations.