From 83ac099b2d4ad0c9f036458f072bb94acb8da7a2 Mon Sep 17 00:00:00 2001
From: Nathan VanBenschoten <nvanbenschoten@gmail.com>
Date: Fri, 24 Jun 2022 20:45:33 -0400
Subject: [PATCH] kv: fix conflict resolution for high-priority, non-txn'al
 requests

Fixes #83342.

This commit reworks the behavior of non-transactional requests in lock
wait-queues when the lock holder or the lock queue waiter has an extreme
priority (min or max priority). In such cases, we allow the lock queue
waiter to immediately push the lock holder out of its way, either by
moving its timestamp to resolve a read-write conflict or aborting it to
resolve a write-write conflict.

This handling was broken in two ways for non-transactional requests.
1. these requests' priorities were not consulted when deciding whether
   to immediately push instead of temporarily delaying while waiting in the
   lock wait-queue. This meant that a high-priority, non-txn request might
   still wait for 50ms (kv.lock_table.coordinator_liveness_push_delay)
   before pushing a lower priority lock holder out of its way.
2. worse, it was possible that if these requests were not in the front
   of a lock wait-queue, they might never push. This was because we had
   logic that disabled a push if it was not needed for the purposes of
   checking liveness, detecting deadlocks, or enforcing timeouts.

This commit resolves both of these issues. It also improves the testing
of transaction priorities in the `kv/kvserver/concurrency` package.
Finally, it consolidates the determination of when a pusher should be
able to push/abort a pushee into a single location.

Release note (bug fix): a bug in transaction conflict resolution which
could allow backups to wait on long-running transactions has been
resolved.
---
 pkg/kv/kvserver/batcheval/cmd_push_txn.go     |   2 +-
 pkg/kv/kvserver/batcheval/transaction.go      |   7 -
 .../concurrency/concurrency_control.go        |   2 +-
 .../concurrency/concurrency_manager_test.go   |  63 +-
 .../concurrency/datadriven_util_test.go       |  35 ++
 .../kvserver/concurrency/lock_table_waiter.go |  36 +-
 .../concurrency/lock_table_waiter_test.go     |   4 +-
 .../testdata/concurrency_manager/priority     | 575 ++++++++++++++++++
 pkg/kv/kvserver/replica_send.go               |   2 +-
 pkg/kv/kvserver/txnwait/queue.go              |  14 +-
 10 files changed, 695 insertions(+), 45 deletions(-)
 create mode 100644 pkg/kv/kvserver/concurrency/testdata/concurrency_manager/priority

diff --git a/pkg/kv/kvserver/batcheval/cmd_push_txn.go b/pkg/kv/kvserver/batcheval/cmd_push_txn.go
index 8134aa90d086..cc99f4c72281 100644
--- a/pkg/kv/kvserver/batcheval/cmd_push_txn.go
+++ b/pkg/kv/kvserver/batcheval/cmd_push_txn.go
@@ -255,7 +255,7 @@ func PushTxn(
 		// If just attempting to cleanup old or already-committed txns,
 		// pusher always fails.
 		pusherWins = false
-	case CanPushWithPriority(&args.PusherTxn, &reply.PusheeTxn):
+	case txnwait.CanPushWithPriority(args.PusherTxn.Priority, reply.PusheeTxn.Priority):
 		reason = "pusher has priority"
 		pusherWins = true
 	case args.Force:
diff --git a/pkg/kv/kvserver/batcheval/transaction.go b/pkg/kv/kvserver/batcheval/transaction.go
index c803ed1a3dd9..bd374efda956 100644
--- a/pkg/kv/kvserver/batcheval/transaction.go
+++ b/pkg/kv/kvserver/batcheval/transaction.go
@@ -118,13 +118,6 @@ func UpdateAbortSpan(
 	return rec.AbortSpan().Put(ctx, readWriter, ms, txn.ID, &curEntry)
 }
 
-// CanPushWithPriority returns true if the given pusher can push the pushee
-// based on its priority.
-func CanPushWithPriority(pusher, pushee *roachpb.Transaction) bool {
-	return (pusher.Priority > enginepb.MinTxnPriority && pushee.Priority == enginepb.MinTxnPriority) ||
-		(pusher.Priority == enginepb.MaxTxnPriority && pushee.Priority < pusher.Priority)
-}
-
 // CanCreateTxnRecord determines whether a transaction record can be created for
 // the provided transaction. If not, the function will return an error. If so,
 // the function may modify the provided transaction.
diff --git a/pkg/kv/kvserver/concurrency/concurrency_control.go b/pkg/kv/kvserver/concurrency/concurrency_control.go
index a3fe9054a10a..ee640416c7a3 100644
--- a/pkg/kv/kvserver/concurrency/concurrency_control.go
+++ b/pkg/kv/kvserver/concurrency/concurrency_control.go
@@ -380,7 +380,7 @@ type Request struct {
 	Timestamp hlc.Timestamp
 
 	// The priority of the request. Only set if Txn is nil.
-	Priority roachpb.UserPriority
+	NonTxnPriority roachpb.UserPriority
 
 	// The consistency level of the request. Only set if Txn is nil.
 	ReadConsistency roachpb.ReadConsistencyType
diff --git a/pkg/kv/kvserver/concurrency/concurrency_manager_test.go b/pkg/kv/kvserver/concurrency/concurrency_manager_test.go
index 24d141aa26e0..86f817b5797c 100644
--- a/pkg/kv/kvserver/concurrency/concurrency_manager_test.go
+++ b/pkg/kv/kvserver/concurrency/concurrency_manager_test.go
@@ -55,7 +55,7 @@ import (
 //
 // The input files use the following DSL:
 //
-// new-txn      name=<txn-name> ts=<int>[,<int>] epoch=<int> [uncertainty-limit=<int>[,<int>]]
+// new-txn      name=<txn-name> ts=<int>[,<int>] [epoch=<int>] [priority] [uncertainty-limit=<int>[,<int>]]
 // new-request  name=<req-name> txn=<txn-name>|none ts=<int>[,<int>] [priority] [inconsistent] [wait-policy=<policy>] [lock-timeout] [max-lock-wait-queue-length=<int>] [poison-policy=[err|wait]]
 //   <proto-name> [<field-name>=<field-value>...] (hint: see scanSingleRequest)
 // sequence     req=<req-name> [eval-kind=<pess|opt|pess-after-opt]
@@ -104,8 +104,12 @@ func TestConcurrencyManagerBasic(t *testing.T) {
 				d.ScanArgs(t, "name", &txnName)
 				ts := scanTimestamp(t, d)
 
-				var epoch int
-				d.ScanArgs(t, "epoch", &epoch)
+				epoch := 0
+				if d.HasArg("epoch") {
+					d.ScanArgs(t, "epoch", &epoch)
+				}
+
+				priority := scanTxnPriority(t, d)
 
 				uncertaintyLimit := ts
 				if d.HasArg("uncertainty-limit") {
@@ -125,7 +129,7 @@ func TestConcurrencyManagerBasic(t *testing.T) {
 						Epoch:          enginepb.TxnEpoch(epoch),
 						WriteTimestamp: ts,
 						MinTimestamp:   ts,
-						Priority:       1, // not min or max
+						Priority:       priority,
 					},
 					ReadTimestamp:          ts,
 					GlobalUncertaintyLimit: uncertaintyLimit,
@@ -159,6 +163,7 @@ func TestConcurrencyManagerBasic(t *testing.T) {
 					readConsistency = roachpb.INCONSISTENT
 				}
 
+				priority := scanUserPriority(t, d)
 				waitPolicy := scanWaitPolicy(t, d, false /* required */)
 
 				var lockTimeout time.Duration
@@ -182,9 +187,9 @@ func TestConcurrencyManagerBasic(t *testing.T) {
 				latchSpans, lockSpans := c.collectSpans(t, txn, ts, reqs)
 
 				c.requestsByName[reqName] = concurrency.Request{
-					Txn:       txn,
-					Timestamp: ts,
-					// TODO(nvanbenschoten): test Priority
+					Txn:                    txn,
+					Timestamp:              ts,
+					NonTxnPriority:         priority,
 					ReadConsistency:        readConsistency,
 					WaitPolicy:             waitPolicy,
 					LockTimeout:            lockTimeout,
@@ -693,19 +698,47 @@ func (c *cluster) PushTransaction(
 		}
 		defer c.unregisterPush(push)
 	}
+	var pusherPriority enginepb.TxnPriority
+	if h.Txn != nil {
+		pusherPriority = h.Txn.Priority
+	} else {
+		pusherPriority = roachpb.MakePriority(h.UserPriority)
+	}
+	pushTo := h.Timestamp.Next()
 	for {
 		// Is the pushee pushed?
 		pusheeTxn, pusheeRecordSig := pusheeRecord.asTxn()
-		var pushed bool
-		switch pushType {
-		case roachpb.PUSH_TIMESTAMP:
-			pushed = h.Timestamp.Less(pusheeTxn.WriteTimestamp) || pusheeTxn.Status.IsFinalized()
-		case roachpb.PUSH_ABORT, roachpb.PUSH_TOUCH:
-			pushed = pusheeTxn.Status.IsFinalized()
+		// NOTE: this logic is adapted from cmd_push_txn.go.
+		var pusherWins bool
+		switch {
+		case pusheeTxn.Status.IsFinalized():
+			// Already finalized.
+			return pusheeTxn, nil
+		case pushType == roachpb.PUSH_TIMESTAMP && pushTo.LessEq(pusheeTxn.WriteTimestamp):
+			// Already pushed.
+			return pusheeTxn, nil
+		case pushType == roachpb.PUSH_TOUCH:
+			pusherWins = false
+		case txnwait.CanPushWithPriority(pusherPriority, pusheeTxn.Priority):
+			pusherWins = true
 		default:
-			return nil, roachpb.NewErrorf("unexpected push type: %s", pushType)
+			pusherWins = false
 		}
-		if pushed {
+		if pusherWins {
+			switch pushType {
+			case roachpb.PUSH_ABORT:
+				log.Eventf(ctx, "pusher aborted pushee")
+				err = c.updateTxnRecord(pusheeTxn.ID, roachpb.ABORTED, pusheeTxn.WriteTimestamp)
+			case roachpb.PUSH_TIMESTAMP:
+				log.Eventf(ctx, "pusher pushed pushee to %s", pushTo)
+				err = c.updateTxnRecord(pusheeTxn.ID, pusheeTxn.Status, pushTo)
+			default:
+				err = errors.Errorf("unexpected push type: %s", pushType)
+			}
+			if err != nil {
+				return nil, roachpb.NewError(err)
+			}
+			pusheeTxn, _ = pusheeRecord.asTxn()
 			return pusheeTxn, nil
 		}
 		// If PUSH_TOUCH, return error instead of waiting.
diff --git a/pkg/kv/kvserver/concurrency/datadriven_util_test.go b/pkg/kv/kvserver/concurrency/datadriven_util_test.go
index a8a477e7eb1f..920ad9665cf1 100644
--- a/pkg/kv/kvserver/concurrency/datadriven_util_test.go
+++ b/pkg/kv/kvserver/concurrency/datadriven_util_test.go
@@ -44,6 +44,41 @@ func scanTimestampWithName(t *testing.T, d *datadriven.TestData, name string) hl
 	return ts
 }
 
+func scanTxnPriority(t *testing.T, d *datadriven.TestData) enginepb.TxnPriority {
+	priority := scanUserPriority(t, d)
+	// NB: don't use roachpb.MakePriority to avoid randomness.
+	switch priority {
+	case roachpb.MinUserPriority:
+		return enginepb.MinTxnPriority
+	case roachpb.NormalUserPriority:
+		return 1 // not min nor max
+	case roachpb.MaxUserPriority:
+		return enginepb.MaxTxnPriority
+	default:
+		d.Fatalf(t, "unknown priority: %s", priority)
+		return 0
+	}
+}
+
+func scanUserPriority(t *testing.T, d *datadriven.TestData) roachpb.UserPriority {
+	const key = "priority"
+	priS := "normal"
+	if d.HasArg(key) {
+		d.ScanArgs(t, key, &priS)
+	}
+	switch priS {
+	case "low":
+		return roachpb.MinUserPriority
+	case "normal":
+		return roachpb.NormalUserPriority
+	case "high":
+		return roachpb.MaxUserPriority
+	default:
+		d.Fatalf(t, "unknown priority: %s", priS)
+		return 0
+	}
+}
+
 func scanLockStrength(t *testing.T, d *datadriven.TestData) lock.Strength {
 	var strS string
 	d.ScanArgs(t, "strength", &strS)
diff --git a/pkg/kv/kvserver/concurrency/lock_table_waiter.go b/pkg/kv/kvserver/concurrency/lock_table_waiter.go
index 8aacb3941863..e72897bd559c 100644
--- a/pkg/kv/kvserver/concurrency/lock_table_waiter.go
+++ b/pkg/kv/kvserver/concurrency/lock_table_waiter.go
@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/intentresolver"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/txnwait"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
@@ -221,9 +222,14 @@ func (w *lockTableWaiterImpl) WaitOn(
 				// still active.
 				timeoutPush := req.LockTimeout != 0
 
+				// If the pushee has the minimum priority or if the pusher has the
+				// maximum priority, push immediately to proceed without queueing.
+				// The push should succeed without entering the txn wait-queue.
+				priorityPush := canPushWithPriority(req, state)
+
 				// If the request doesn't want to perform a delayed push for any
 				// reason, continue waiting without a timer.
-				if !livenessPush && !deadlockPush && !timeoutPush {
+				if !(livenessPush || deadlockPush || timeoutPush || priorityPush) {
 					continue
 				}
 
@@ -250,12 +256,7 @@ func (w *lockTableWaiterImpl) WaitOn(
 					}
 					delay = minDuration(delay, w.timeUntilDeadline(lockDeadline))
 				}
-
-				// However, if the pushee has the minimum priority or if the
-				// pusher has the maximum priority, push immediately.
-				// TODO(nvanbenschoten): flesh these interactions out more and
-				// add some testing.
-				if hasMinPriority(state.txn) || hasMaxPriority(req.Txn) {
+				if priorityPush {
 					delay = 0
 				}
 
@@ -754,7 +755,7 @@ func (w *lockTableWaiterImpl) pushRequestTxn(
 func (w *lockTableWaiterImpl) pushHeader(req Request) roachpb.Header {
 	h := roachpb.Header{
 		Timestamp:    req.Timestamp,
-		UserPriority: req.Priority,
+		UserPriority: req.NonTxnPriority,
 	}
 	if req.Txn != nil {
 		// We are going to hand the header (and thus the transaction proto) to
@@ -1224,12 +1225,19 @@ func newWriteIntentErr(
 	return err
 }
 
-func hasMinPriority(txn *enginepb.TxnMeta) bool {
-	return txn != nil && txn.Priority == enginepb.MinTxnPriority
-}
-
-func hasMaxPriority(txn *roachpb.Transaction) bool {
-	return txn != nil && txn.Priority == enginepb.MaxTxnPriority
+func canPushWithPriority(req Request, s waitingState) bool {
+	var pusher, pushee enginepb.TxnPriority
+	if req.Txn != nil {
+		pusher = req.Txn.Priority
+	} else {
+		pusher = roachpb.MakePriority(req.NonTxnPriority)
+	}
+	if s.txn == nil {
+		// Can't push a non-transactional request.
+		return false
+	}
+	pushee = s.txn.Priority
+	return txnwait.CanPushWithPriority(pusher, pushee)
 }
 
 func minDuration(a, b time.Duration) time.Duration {
diff --git a/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go b/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go
index e498a078c2e5..1beb04b0f7fc 100644
--- a/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go
+++ b/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go
@@ -241,8 +241,8 @@ func TestLockTableWaiterWithNonTxn(t *testing.T) {
 	reqHeaderTS := hlc.Timestamp{WallTime: 10}
 	makeReq := func() Request {
 		return Request{
-			Timestamp: reqHeaderTS,
-			Priority:  roachpb.NormalUserPriority,
+			Timestamp:      reqHeaderTS,
+			NonTxnPriority: roachpb.NormalUserPriority,
 		}
 	}
 
diff --git a/pkg/kv/kvserver/concurrency/testdata/concurrency_manager/priority b/pkg/kv/kvserver/concurrency/testdata/concurrency_manager/priority
new file mode 100644
index 000000000000..d15288317477
--- /dev/null
+++ b/pkg/kv/kvserver/concurrency/testdata/concurrency_manager/priority
@@ -0,0 +1,575 @@
+new-txn name=txnLowPushee ts=10,1 priority=low
+----
+
+new-txn name=txnNormalPushee ts=10,1 priority=normal
+----
+
+new-txn name=txnHighPushee ts=10,1 priority=high
+----
+
+new-txn name=txnLowPusher1 ts=10,1 priority=low
+----
+
+new-txn name=txnLowPusher2 ts=10,1 priority=low
+----
+
+new-txn name=txnLowPusher3 ts=10,1 priority=low
+----
+
+new-txn name=txnNormalPusher1 ts=10,1 priority=normal
+----
+
+new-txn name=txnNormalPusher2 ts=10,1 priority=normal
+----
+
+new-txn name=txnHighPusher ts=10,1 priority=high
+----
+
+# -------------------------------------------------------------
+# Prep: "Pushee" txns acquire 2 locks each.
+# -------------------------------------------------------------
+
+new-request name=req1 txn=txnLowPushee ts=10,1
+  put key=kLow1  value=v
+  put key=kLow2  value=v
+----
+
+sequence req=req1
+----
+[1] sequence req1: sequencing request
+[1] sequence req1: acquiring latches
+[1] sequence req1: scanning lock table for conflicting locks
+[1] sequence req1: sequencing complete, returned guard
+
+on-lock-acquired req=req1 key=kLow1
+----
+[-] acquire lock: txn 00000001 @ kLow1
+
+on-lock-acquired req=req1 key=kLow2
+----
+[-] acquire lock: txn 00000001 @ kLow2
+
+finish req=req1
+----
+[-] finish req1: finishing request
+
+new-request name=req2 txn=txnNormalPushee ts=10,1
+  put key=kNormal1  value=v
+  put key=kNormal2  value=v
+----
+
+sequence req=req2
+----
+[2] sequence req2: sequencing request
+[2] sequence req2: acquiring latches
+[2] sequence req2: scanning lock table for conflicting locks
+[2] sequence req2: sequencing complete, returned guard
+
+on-lock-acquired req=req2 key=kNormal1
+----
+[-] acquire lock: txn 00000002 @ kNormal1
+
+on-lock-acquired req=req2 key=kNormal2
+----
+[-] acquire lock: txn 00000002 @ kNormal2
+
+finish req=req2
+----
+[-] finish req2: finishing request
+
+new-request name=req3 txn=txnHighPushee ts=10,1
+  put key=kHigh1  value=v
+  put key=kHigh2  value=v
+----
+
+sequence req=req3
+----
+[3] sequence req3: sequencing request
+[3] sequence req3: acquiring latches
+[3] sequence req3: scanning lock table for conflicting locks
+[3] sequence req3: sequencing complete, returned guard
+
+on-lock-acquired req=req3 key=kHigh1
+----
+[-] acquire lock: txn 00000003 @ kHigh1
+
+on-lock-acquired req=req3 key=kHigh2
+----
+[-] acquire lock: txn 00000003 @ kHigh2
+
+finish req=req3
+----
+[-] finish req3: finishing request
+
+debug-lock-table
+----
+global: num=6
+ lock: "kHigh1"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kHigh2"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow1"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow2"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kNormal1"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kNormal2"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+local: num=0
+
+# -------------------------------------------------------------
+# Push (timestamp) the low priority txn using:
+# - a low priority transactional request
+# - a normal priority transactional request
+# -------------------------------------------------------------
+
+new-request name=req4 txn=txnLowPusher1 ts=10,1
+  get key=kLow1
+----
+
+new-request name=req5 txn=txnNormalPusher1 ts=10,1
+  get key=kLow1
+----
+
+sequence req=req4
+----
+[4] sequence req4: sequencing request
+[4] sequence req4: acquiring latches
+[4] sequence req4: scanning lock table for conflicting locks
+[4] sequence req4: waiting in lock wait-queues
+[4] sequence req4: lock wait-queue event: wait for (distinguished) txn 00000001 holding lock @ key "kLow1" (queuedWriters: 0, queuedReaders: 1)
+[4] sequence req4: pushing timestamp of txn 00000001 above 10.000000000,1
+[4] sequence req4: blocked on select in concurrency_test.(*cluster).PushTransaction
+
+sequence req=req5
+----
+[4] sequence req4: resolving intent "kLow1" for txn 00000001 with PENDING status and clock observation {1 123.000000000,1}
+[4] sequence req4: lock wait-queue event: done waiting
+[4] sequence req4: conflicted with 00000001-0000-0000-0000-000000000000 on "kLow1" for 0.000s
+[4] sequence req4: acquiring latches
+[4] sequence req4: scanning lock table for conflicting locks
+[4] sequence req4: sequencing complete, returned guard
+[5] sequence req5: sequencing request
+[5] sequence req5: acquiring latches
+[5] sequence req5: scanning lock table for conflicting locks
+[5] sequence req5: waiting in lock wait-queues
+[5] sequence req5: lock wait-queue event: wait for txn 00000001 holding lock @ key "kLow1" (queuedWriters: 0, queuedReaders: 2)
+[5] sequence req5: pushing timestamp of txn 00000001 above 10.000000000,1
+[5] sequence req5: pusher pushed pushee to 10.000000000,2
+[5] sequence req5: resolving intent "kLow1" for txn 00000001 with PENDING status and clock observation {1 123.000000000,3}
+[5] sequence req5: lock wait-queue event: done waiting
+[5] sequence req5: conflicted with 00000001-0000-0000-0000-000000000000 on "kLow1" for 0.000s
+[5] sequence req5: acquiring latches
+[5] sequence req5: scanning lock table for conflicting locks
+[5] sequence req5: sequencing complete, returned guard
+
+finish req=req4
+----
+[-] finish req4: finishing request
+
+finish req=req5
+----
+[-] finish req5: finishing request
+
+debug-lock-table
+----
+global: num=6
+ lock: "kHigh1"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kHigh2"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow1"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow2"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kNormal1"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kNormal2"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+local: num=0
+
+# -------------------------------------------------------------
+# Push (abort) the low priority txn using:
+# - a low priority transactional request
+# - a normal priority non-transactional request
+# -------------------------------------------------------------
+
+new-request name=req6 txn=txnLowPusher1 ts=10,1
+  put key=kLow2  value=v
+----
+
+new-request name=req7 txn=none ts=10,1 priority=normal
+  put key=kLow2  value=v
+----
+
+sequence req=req6
+----
+[6] sequence req6: sequencing request
+[6] sequence req6: acquiring latches
+[6] sequence req6: scanning lock table for conflicting locks
+[6] sequence req6: waiting in lock wait-queues
+[6] sequence req6: lock wait-queue event: wait for (distinguished) txn 00000001 holding lock @ key "kLow2" (queuedWriters: 1, queuedReaders: 0)
+[6] sequence req6: pushing txn 00000001 to abort
+[6] sequence req6: blocked on select in concurrency_test.(*cluster).PushTransaction
+
+sequence req=req7
+----
+[6] sequence req6: resolving intent "kLow2" for txn 00000001 with ABORTED status
+[6] sequence req6: lock wait-queue event: done waiting
+[6] sequence req6: conflicted with 00000001-0000-0000-0000-000000000000 on "kLow2" for 0.000s
+[6] sequence req6: acquiring latches
+[6] sequence req6: scanning lock table for conflicting locks
+[6] sequence req6: sequencing complete, returned guard
+[7] sequence req7: sequencing request
+[7] sequence req7: acquiring latches
+[7] sequence req7: scanning lock table for conflicting locks
+[7] sequence req7: waiting in lock wait-queues
+[7] sequence req7: lock wait-queue event: wait for txn 00000001 holding lock @ key "kLow2" (queuedWriters: 2, queuedReaders: 0)
+[7] sequence req7: pushing txn 00000001 to abort
+[7] sequence req7: pusher aborted pushee
+[7] sequence req7: resolving intent "kLow2" for txn 00000001 with ABORTED status
+[7] sequence req7: lock wait-queue event: wait for (distinguished) txn 00000004 running request @ key "kLow2" (queuedWriters: 1, queuedReaders: 0)
+[7] sequence req7: conflicted with 00000001-0000-0000-0000-000000000000 on "kLow2" for 0.000s
+[7] sequence req7: pushing txn 00000004 to detect request deadlock
+[7] sequence req7: pusher aborted pushee
+[7] sequence req7: blocked on select in concurrency.(*lockTableWaiterImpl).WaitOn
+
+finish req=req6
+----
+[-] finish req6: finishing request
+[7] sequence req7: lock wait-queue event: done waiting
+[7] sequence req7: conflicted with 00000004-0000-0000-0000-000000000000 on "kLow2" for 0.000s
+[7] sequence req7: acquiring latches
+[7] sequence req7: scanning lock table for conflicting locks
+[7] sequence req7: sequencing complete, returned guard
+
+finish req=req7
+----
+[-] finish req7: finishing request
+
+debug-lock-table
+----
+global: num=5
+ lock: "kHigh1"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kHigh2"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow1"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+ lock: "kNormal1"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kNormal2"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+local: num=0
+
+# -------------------------------------------------------------
+# Push (timestamp) the normal priority txn using:
+# - a normal priority transactional request
+# - a high priority transactional request
+# -------------------------------------------------------------
+
+new-request name=req8 txn=txnNormalPusher1 ts=10,1
+  get key=kNormal1
+----
+
+new-request name=req9 txn=txnHighPusher ts=10,1
+  get key=kNormal1
+----
+
+sequence req=req8
+----
+[8] sequence req8: sequencing request
+[8] sequence req8: acquiring latches
+[8] sequence req8: scanning lock table for conflicting locks
+[8] sequence req8: waiting in lock wait-queues
+[8] sequence req8: lock wait-queue event: wait for (distinguished) txn 00000002 holding lock @ key "kNormal1" (queuedWriters: 0, queuedReaders: 1)
+[8] sequence req8: pushing timestamp of txn 00000002 above 10.000000000,1
+[8] sequence req8: blocked on select in concurrency_test.(*cluster).PushTransaction
+
+sequence req=req9
+----
+[8] sequence req8: resolving intent "kNormal1" for txn 00000002 with PENDING status and clock observation {1 123.000000000,6}
+[8] sequence req8: lock wait-queue event: done waiting
+[8] sequence req8: conflicted with 00000002-0000-0000-0000-000000000000 on "kNormal1" for 0.000s
+[8] sequence req8: acquiring latches
+[8] sequence req8: scanning lock table for conflicting locks
+[8] sequence req8: sequencing complete, returned guard
+[9] sequence req9: sequencing request
+[9] sequence req9: acquiring latches
+[9] sequence req9: scanning lock table for conflicting locks
+[9] sequence req9: waiting in lock wait-queues
+[9] sequence req9: lock wait-queue event: wait for txn 00000002 holding lock @ key "kNormal1" (queuedWriters: 0, queuedReaders: 2)
+[9] sequence req9: pushing timestamp of txn 00000002 above 10.000000000,1
+[9] sequence req9: pusher pushed pushee to 10.000000000,2
+[9] sequence req9: resolving intent "kNormal1" for txn 00000002 with PENDING status and clock observation {1 123.000000000,8}
+[9] sequence req9: lock wait-queue event: done waiting
+[9] sequence req9: conflicted with 00000002-0000-0000-0000-000000000000 on "kNormal1" for 0.000s
+[9] sequence req9: acquiring latches
+[9] sequence req9: scanning lock table for conflicting locks
+[9] sequence req9: sequencing complete, returned guard
+
+finish req=req8
+----
+[-] finish req8: finishing request
+
+finish req=req9
+----
+[-] finish req9: finishing request
+
+debug-lock-table
+----
+global: num=5
+ lock: "kHigh1"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kHigh2"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow1"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+ lock: "kNormal1"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl epoch: 0, seqs: [0]
+ lock: "kNormal2"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+local: num=0
+
+# -------------------------------------------------------------
+# Push (abort) the normal priority txn using:
+# - a normal priority transactional request
+# - a high priority non-transactional request
+# -------------------------------------------------------------
+
+new-request name=req10 txn=txnNormalPusher1 ts=10,1
+  put key=kNormal2  value=v
+----
+
+new-request name=req11 txn=none ts=10,1 priority=high
+  put key=kNormal2  value=v
+----
+
+sequence req=req10
+----
+[10] sequence req10: sequencing request
+[10] sequence req10: acquiring latches
+[10] sequence req10: scanning lock table for conflicting locks
+[10] sequence req10: waiting in lock wait-queues
+[10] sequence req10: lock wait-queue event: wait for (distinguished) txn 00000002 holding lock @ key "kNormal2" (queuedWriters: 1, queuedReaders: 0)
+[10] sequence req10: pushing txn 00000002 to abort
+[10] sequence req10: blocked on select in concurrency_test.(*cluster).PushTransaction
+
+sequence req=req11
+----
+[10] sequence req10: resolving intent "kNormal2" for txn 00000002 with ABORTED status
+[10] sequence req10: lock wait-queue event: done waiting
+[10] sequence req10: conflicted with 00000002-0000-0000-0000-000000000000 on "kNormal2" for 0.000s
+[10] sequence req10: acquiring latches
+[10] sequence req10: scanning lock table for conflicting locks
+[10] sequence req10: sequencing complete, returned guard
+[11] sequence req11: sequencing request
+[11] sequence req11: acquiring latches
+[11] sequence req11: scanning lock table for conflicting locks
+[11] sequence req11: waiting in lock wait-queues
+[11] sequence req11: lock wait-queue event: wait for txn 00000002 holding lock @ key "kNormal2" (queuedWriters: 2, queuedReaders: 0)
+[11] sequence req11: pushing txn 00000002 to abort
+[11] sequence req11: pusher aborted pushee
+[11] sequence req11: resolving intent "kNormal2" for txn 00000002 with ABORTED status
+[11] sequence req11: lock wait-queue event: wait for (distinguished) txn 00000007 running request @ key "kNormal2" (queuedWriters: 1, queuedReaders: 0)
+[11] sequence req11: conflicted with 00000002-0000-0000-0000-000000000000 on "kNormal2" for 0.000s
+[11] sequence req11: pushing txn 00000007 to detect request deadlock
+[11] sequence req11: pusher aborted pushee
+[11] sequence req11: blocked on select in concurrency.(*lockTableWaiterImpl).WaitOn
+
+finish req=req10
+----
+[-] finish req10: finishing request
+[11] sequence req11: lock wait-queue event: done waiting
+[11] sequence req11: conflicted with 00000007-0000-0000-0000-000000000000 on "kNormal2" for 0.000s
+[11] sequence req11: acquiring latches
+[11] sequence req11: scanning lock table for conflicting locks
+[11] sequence req11: sequencing complete, returned guard
+
+finish req=req11
+----
+[-] finish req11: finishing request
+
+debug-lock-table
+----
+global: num=4
+ lock: "kHigh1"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kHigh2"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow1"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+ lock: "kNormal1"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+local: num=0
+
+# -------------------------------------------------------------
+# Push (timestamp) the high priority txn using:
+# - a normal priority transactional request
+# - a high priority transactional request
+# -------------------------------------------------------------
+
+new-request name=req12 txn=txnNormalPusher2 ts=10,1
+  get key=kHigh1
+----
+
+new-request name=req13 txn=txnHighPusher ts=10,1
+  get key=kHigh1
+----
+
+sequence req=req12
+----
+[12] sequence req12: sequencing request
+[12] sequence req12: acquiring latches
+[12] sequence req12: scanning lock table for conflicting locks
+[12] sequence req12: waiting in lock wait-queues
+[12] sequence req12: lock wait-queue event: wait for (distinguished) txn 00000003 holding lock @ key "kHigh1" (queuedWriters: 0, queuedReaders: 1)
+[12] sequence req12: pushing timestamp of txn 00000003 above 10.000000000,1
+[12] sequence req12: blocked on select in concurrency_test.(*cluster).PushTransaction
+
+sequence req=req13
+----
+[13] sequence req13: sequencing request
+[13] sequence req13: acquiring latches
+[13] sequence req13: scanning lock table for conflicting locks
+[13] sequence req13: waiting in lock wait-queues
+[13] sequence req13: lock wait-queue event: wait for txn 00000003 holding lock @ key "kHigh1" (queuedWriters: 0, queuedReaders: 2)
+[13] sequence req13: pushing timestamp of txn 00000003 above 10.000000000,1
+[13] sequence req13: blocked on select in concurrency_test.(*cluster).PushTransaction
+
+on-txn-updated txn=txnHighPushee status=pending ts=10,2
+----
+[-] update txn: increasing timestamp of txnHighPushee
+[12] sequence req12: resolving intent "kHigh1" for txn 00000003 with PENDING status and clock observation {1 123.000000000,11}
+[12] sequence req12: lock wait-queue event: done waiting
+[12] sequence req12: conflicted with 00000003-0000-0000-0000-000000000000 on "kHigh1" for 0.000s
+[12] sequence req12: acquiring latches
+[12] sequence req12: scanning lock table for conflicting locks
+[12] sequence req12: sequencing complete, returned guard
+[13] sequence req13: resolving intent "kHigh1" for txn 00000003 with PENDING status and clock observation {1 123.000000000,13}
+[13] sequence req13: lock wait-queue event: done waiting
+[13] sequence req13: conflicted with 00000003-0000-0000-0000-000000000000 on "kHigh1" for 0.000s
+[13] sequence req13: acquiring latches
+[13] sequence req13: scanning lock table for conflicting locks
+[13] sequence req13: sequencing complete, returned guard
+
+finish req=req12
+----
+[-] finish req12: finishing request
+
+finish req=req13
+----
+[-] finish req13: finishing request
+
+debug-lock-table
+----
+global: num=4
+ lock: "kHigh1"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl epoch: 0, seqs: [0]
+ lock: "kHigh2"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,1, info: unrepl epoch: 0, seqs: [0]
+ lock: "kLow1"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+ lock: "kNormal1"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+local: num=0
+
+# -------------------------------------------------------------
+# Push (abort) the high priority txn using:
+# - a normal priority transactional request
+# - a high priority non-transactional request
+# -------------------------------------------------------------
+
+new-request name=req14 txn=txnNormalPusher2 ts=10,1
+  put key=kHigh2  value=v
+----
+
+new-request name=req15 txn=none ts=10,1 priority=high
+  put key=kHigh2  value=v
+----
+
+sequence req=req14
+----
+[14] sequence req14: sequencing request
+[14] sequence req14: acquiring latches
+[14] sequence req14: scanning lock table for conflicting locks
+[14] sequence req14: waiting in lock wait-queues
+[14] sequence req14: lock wait-queue event: wait for (distinguished) txn 00000003 holding lock @ key "kHigh2" (queuedWriters: 1, queuedReaders: 0)
+[14] sequence req14: pushing txn 00000003 to abort
+[14] sequence req14: blocked on select in concurrency_test.(*cluster).PushTransaction
+
+sequence req=req15
+----
+[15] sequence req15: sequencing request
+[15] sequence req15: acquiring latches
+[15] sequence req15: scanning lock table for conflicting locks
+[15] sequence req15: waiting in lock wait-queues
+[15] sequence req15: lock wait-queue event: wait for txn 00000003 holding lock @ key "kHigh2" (queuedWriters: 2, queuedReaders: 0)
+[15] sequence req15: blocked on select in concurrency.(*lockTableWaiterImpl).WaitOn
+
+on-txn-updated txn=txnHighPushee status=committed
+----
+[-] update txn: committing txnHighPushee
+[14] sequence req14: resolving intent "kHigh2" for txn 00000003 with COMMITTED status
+[14] sequence req14: lock wait-queue event: done waiting
+[14] sequence req14: conflicted with 00000003-0000-0000-0000-000000000000 on "kHigh2" for 0.000s
+[14] sequence req14: acquiring latches
+[14] sequence req14: scanning lock table for conflicting locks
+[14] sequence req14: sequencing complete, returned guard
+[15] sequence req15: lock wait-queue event: wait for (distinguished) txn 00000008 running request @ key "kHigh2" (queuedWriters: 1, queuedReaders: 0)
+[15] sequence req15: conflicted with 00000003-0000-0000-0000-000000000000 on "kHigh2" for 0.000s
+[15] sequence req15: pushing txn 00000008 to detect request deadlock
+[15] sequence req15: pusher aborted pushee
+[15] sequence req15: blocked on select in concurrency.(*lockTableWaiterImpl).WaitOn
+
+finish req=req14
+----
+[-] finish req14: finishing request
+[15] sequence req15: lock wait-queue event: done waiting
+[15] sequence req15: conflicted with 00000008-0000-0000-0000-000000000000 on "kHigh2" for 0.000s
+[15] sequence req15: acquiring latches
+[15] sequence req15: scanning lock table for conflicting locks
+[15] sequence req15: sequencing complete, returned guard
+
+finish req=req15
+----
+[-] finish req15: finishing request
+
+debug-lock-table
+----
+global: num=3
+ lock: "kHigh1"
+  holder: txn: 00000003-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: committed] epoch: 0, seqs: [0]
+ lock: "kLow1"
+  holder: txn: 00000001-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+ lock: "kNormal1"
+  holder: txn: 00000002-0000-0000-0000-000000000000, ts: 10.000000000,2, info: unrepl [holder finalized: aborted] epoch: 0, seqs: [0]
+local: num=0
+
+# -------------------------------------------------------------
+# Scan across keyspace to clear out all aborted locks.
+# -------------------------------------------------------------
+
+new-request name=req16 txn=none ts=11,1
+  scan key=a endkey=z
+----
+
+sequence req=req16
+----
+[16] sequence req16: sequencing request
+[16] sequence req16: acquiring latches
+[16] sequence req16: scanning lock table for conflicting locks
+[16] sequence req16: sequencing complete, returned guard
+
+finish req=req16
+----
+[-] finish req16: finishing request
+
+debug-lock-table
+----
+global: num=0
+local: num=0
+
+reset
+----
diff --git a/pkg/kv/kvserver/replica_send.go b/pkg/kv/kvserver/replica_send.go
index 6da9ea7851b6..495bc9b64730 100644
--- a/pkg/kv/kvserver/replica_send.go
+++ b/pkg/kv/kvserver/replica_send.go
@@ -445,7 +445,7 @@ func (r *Replica) executeBatchWithConcurrencyRetries(
 		g, resp, pErr = r.concMgr.SequenceReq(ctx, g, concurrency.Request{
 			Txn:             ba.Txn,
 			Timestamp:       ba.Timestamp,
-			Priority:        ba.UserPriority,
+			NonTxnPriority:  ba.UserPriority,
 			ReadConsistency: ba.ReadConsistency,
 			WaitPolicy:      ba.WaitPolicy,
 			LockTimeout:     ba.LockTimeout,
diff --git a/pkg/kv/kvserver/txnwait/queue.go b/pkg/kv/kvserver/txnwait/queue.go
index 12bc01d08677..2c1c1231e80d 100644
--- a/pkg/kv/kvserver/txnwait/queue.go
+++ b/pkg/kv/kvserver/txnwait/queue.go
@@ -69,19 +69,25 @@ func ShouldPushImmediately(req *roachpb.PushTxnRequest) bool {
 	if !(req.PushType == roachpb.PUSH_ABORT || req.PushType == roachpb.PUSH_TIMESTAMP) {
 		return true
 	}
-	p1, p2 := req.PusherTxn.Priority, req.PusheeTxn.Priority
-	if p1 > p2 && (p1 == enginepb.MaxTxnPriority || p2 == enginepb.MinTxnPriority) {
+	if CanPushWithPriority(req.PusherTxn.Priority, req.PusheeTxn.Priority) {
 		return true
 	}
 	return false
 }
 
+// CanPushWithPriority returns true if the given pusher can push the pushee
+// based on its priority.
+func CanPushWithPriority(pusher, pushee enginepb.TxnPriority) bool {
+	return pusher > pushee &&
+		(pusher == enginepb.MaxTxnPriority || pushee == enginepb.MinTxnPriority)
+}
+
 // isPushed returns whether the PushTxn request has already been
 // fulfilled by the current transaction state. This may be true
 // for transactions with pushed timestamps.
 func isPushed(req *roachpb.PushTxnRequest, txn *roachpb.Transaction) bool {
-	return (txn.Status.IsFinalized() ||
-		(req.PushType == roachpb.PUSH_TIMESTAMP && req.PushTo.LessEq(txn.WriteTimestamp)))
+	return txn.Status.IsFinalized() ||
+		(req.PushType == roachpb.PUSH_TIMESTAMP && req.PushTo.LessEq(txn.WriteTimestamp))
 }
 
 // TxnExpiration computes the timestamp after which the transaction will be