kv: add PrevLease check in RequestLease

This commit adds a check that `args.PrevLease` is equivalent to
`cArgs.EvalCtx.GetLease()` to RequestLease. This ensures that the
validation here is consistent with the validation that was performed
when the lease request was constructed.

Release note: None
Epic: None

pkg/kv/kvserver/batcheval/cmd_lease_request.go

@@ -12,6 +12,7 @@ package batcheval
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
@@ -67,6 +68,15 @@ func RequestLease(
 		Requested: args.Lease,
 	}
 
+	// However, we verify that the current lease's sequence number and proposed
+	// timestamp match the provided PrevLease. This ensures that the validation
+	// here is consistent with the validation that was performed when the lease
+	// request was constructed.
+	if prevLease.Sequence != args.PrevLease.Sequence || !prevLease.ProposedTS.Equal(args.PrevLease.ProposedTS) {
+		rErr.Message = fmt.Sprintf("expected previous lease %s, found %s", args.PrevLease, prevLease)
+		return newFailedLeaseTrigger(false /* isTransfer */), rErr
+	}
+
 	// MIGRATION(tschottdorf): needed to apply Raft commands which got proposed
 	// before the StartStasis field was introduced.
 	newLease := args.Lease

pkg/kv/kvserver/replica_test.go

@@ -966,7 +966,8 @@ func TestReplicaLease(t *testing.T) {
 					kvpb.AdmissionHeader{},
 				),
 				Args: &kvpb.RequestLeaseRequest{
-					Lease: lease,
+					Lease:     lease,
+					PrevLease: tc.repl.CurrentLeaseStatus(ctx).Lease,
 				},
 			}, &kvpb.RequestLeaseResponse{}); !testutils.IsError(err, "replica not found") {
 			t.Fatalf("unexpected error: %+v", err)
@@ -1317,7 +1318,7 @@ func TestReplicaLeaseRejectUnknownRaftNodeID(t *testing.T) {
 	st := tc.repl.CurrentLeaseStatus(ctx)
 	ba := &kvpb.BatchRequest{}
 	ba.Timestamp = tc.repl.store.Clock().Now()
-	ba.Add(&kvpb.RequestLeaseRequest{Lease: *lease})
+	ba.Add(&kvpb.RequestLeaseRequest{Lease: *lease, PrevLease: st.Lease})
 	_, tok := tc.repl.mu.proposalBuf.TrackEvaluatingRequest(ctx, hlc.MinTimestamp)
 	ch, _, _, _, pErr := tc.repl.evalAndPropose(ctx, ba, allSpansGuard(), &st, uncertainty.Interval{}, tok.Move(ctx))
 	if pErr == nil {