batcheval: allow sequence number regression on lock acquire

This was an oversight in
39d519b. When flushing the txn write
buffer, we may also be regressing a lock that was previously flushed
added to the replicated lock table via an unreplicated lock table
flush.

Fixes #148789

Epic: none

Release note: None

Author: stevendanna

pkg/kv/kvserver/batcheval/cmd_get.go

@@ -120,8 +120,13 @@ func Get(
 	shouldLockKey := getRes.Value != nil || args.LockNonExisting
 	var res result.Result
 	if args.KeyLockingStrength != lock.None && shouldLockKey {
+		// ExpectExclusionSince is used by callers (namely, txnWriteBuffers) that
+		// are likely to be sending replicated, locking Get requests at sequence
+		// numbers corresponding to unreplicated locks taken earlier in the
+		// transaction. In this case, sequence number regression is not unexpected.
+		allowSequenceNumberRegression := args.ExpectExclusionSince.IsSet()
 		acq, err := acquireLockOnKey(ctx, readWriter, h.Txn, args.KeyLockingStrength,
-			args.KeyLockingDurability, args.Key, cArgs.Stats, cArgs.EvalCtx.ClusterSettings())
+			args.KeyLockingDurability, args.Key, cArgs.Stats, cArgs.EvalCtx.ClusterSettings(), allowSequenceNumberRegression)
 		if err != nil {
 			return result.Result{}, err
 		}

pkg/kv/kvserver/batcheval/lock.go

@@ -137,7 +137,7 @@ func acquireLocksOnKeys(
 	case kvpb.BATCH_RESPONSE:
 		err := storage.MVCCScanDecodeKeyValues(scanRes.KVData, func(key storage.MVCCKey, _ []byte) error {
 			k := copyKey(key.Key)
-			acq, err := acquireLockOnKey(ctx, readWriter, txn, str, dur, k, ms, settings)
+			acq, err := acquireLockOnKey(ctx, readWriter, txn, str, dur, k, ms, settings, false /* 	allowSequenceNumberRegression */)
 			if err != nil {
 				return err
 			}
@@ -153,7 +153,7 @@ func acquireLocksOnKeys(
 	case kvpb.KEY_VALUES:
 		for _, row := range scanRes.KVs {
 			k := copyKey(row.Key)
-			acq, err := acquireLockOnKey(ctx, readWriter, txn, str, dur, k, ms, settings)
+			acq, err := acquireLockOnKey(ctx, readWriter, txn, str, dur, k, ms, settings, false /* 	allowSequenceNumberRegression */)
 			if err != nil {
 				return nil, err
 			}
@@ -198,6 +198,7 @@ func acquireLockOnKey(
 	key roachpb.Key,
 	ms *enginepb.MVCCStats,
 	settings *cluster.Settings,
+	allowSequenceNumberRegression bool,
 ) (roachpb.LockAcquisition, error) {
 	maxLockConflicts := storage.MaxConflictsPerLockConflictError.Get(&settings.SV)
 	targetLockConflictBytes := storage.TargetBytesPerLockConflictError.Get(&settings.SV)
@@ -233,7 +234,8 @@ func acquireLockOnKey(
 		// conflicts with un-contended replicated locks -- we need to do so before
 		// we can acquire our own replicated lock; do that now, and also acquire
 		// the replicated lock if no conflicts are found.
-		if err := storage.MVCCAcquireLock(ctx, readWriter, &txn.TxnMeta, txn.IgnoredSeqNums, str, key, ms, maxLockConflicts, targetLockConflictBytes, false); err != nil {
+		if err := storage.MVCCAcquireLock(ctx, readWriter, &txn.TxnMeta, txn.IgnoredSeqNums, str,
+			key, ms, maxLockConflicts, targetLockConflictBytes, allowSequenceNumberRegression); err != nil {
 			return roachpb.LockAcquisition{}, err
 		}
 	default: