diff --git a/pkg/kv/kvserver/concurrency/concurrency_control.go b/pkg/kv/kvserver/concurrency/concurrency_control.go
index 2fa422045d6d..3c06e064848c 100644
--- a/pkg/kv/kvserver/concurrency/concurrency_control.go
+++ b/pkg/kv/kvserver/concurrency/concurrency_control.go
@@ -592,7 +592,7 @@ type lockTable interface {
 	// lockTableGuard and the subsequent calls reuse the previously returned
 	// one. The latches needed by the request must be held when calling this
 	// function.
-	ScanAndEnqueue(Request, lockTableGuard) lockTableGuard
+	ScanAndEnqueue(Request, lockTableGuard) (lockTableGuard, *Error)
 
 	// ScanOptimistic takes a snapshot of the lock table for later checking for
 	// conflicts, and returns a guard. It is for optimistic evaluation of
@@ -760,7 +760,7 @@ type lockTableGuard interface {
 	NewStateChan() chan struct{}
 
 	// CurState returns the latest waiting state.
-	CurState() waitingState
+	CurState() (waitingState, error)
 
 	// ResolveBeforeScanning lists the locks to resolve before scanning again.
 	// This must be called after:
diff --git a/pkg/kv/kvserver/concurrency/concurrency_manager.go b/pkg/kv/kvserver/concurrency/concurrency_manager.go
index 7fa39598f611..900a014aa6ee 100644
--- a/pkg/kv/kvserver/concurrency/concurrency_manager.go
+++ b/pkg/kv/kvserver/concurrency/concurrency_manager.go
@@ -332,7 +332,10 @@ func (m *managerImpl) sequenceReqWithGuard(
 		} else {
 			// Scan for conflicting locks.
 			log.Event(ctx, "scanning lock table for conflicting locks")
-			g.ltg = m.lt.ScanAndEnqueue(g.Req, g.ltg)
+			g.ltg, err = m.lt.ScanAndEnqueue(g.Req, g.ltg)
+			if err != nil {
+				return nil, err
+			}
 		}
 
 		// Wait on conflicting locks, if necessary. Note that this will never be
diff --git a/pkg/kv/kvserver/concurrency/lock_table.go b/pkg/kv/kvserver/concurrency/lock_table.go
index 78ab1587d359..e1edd6e0e9dd 100644
--- a/pkg/kv/kvserver/concurrency/lock_table.go
+++ b/pkg/kv/kvserver/concurrency/lock_table.go
@@ -568,19 +568,22 @@ func (g *lockTableGuardImpl) NewStateChan() chan struct{} {
 	return g.mu.signal
 }
 
-func (g *lockTableGuardImpl) CurState() waitingState {
+func (g *lockTableGuardImpl) CurState() (waitingState, error) {
 	g.mu.Lock()
 	defer g.mu.Unlock()
 	if !g.mu.mustComputeWaitingState {
-		return g.mu.state
+		return g.mu.state, nil
 	}
 	// Not actively waiting anywhere so no one else can set
 	// mustComputeWaitingState to true while this method executes.
 	g.mu.mustComputeWaitingState = false
 	g.mu.Unlock()
-	g.resumeScan(false /* notify */)
+	err := g.resumeScan(false /* notify */)
 	g.mu.Lock() // Unlock deferred
-	return g.mu.state
+	if err != nil {
+		return waitingState{}, err
+	}
+	return g.mu.state, nil
 }
 
 // updateStateToDoneWaitingLocked updates the request's waiting state to
@@ -700,7 +703,11 @@ func (g *lockTableGuardImpl) IsKeyLockedByConflictingTxn(
 		return false, nil, nil
 	}
 
-	if l.alreadyHoldsLockAndIsAllowedToProceed(g, str) {
+	isAllowedToProceed, err := l.alreadyHoldsLockAndIsAllowedToProceed(g, str)
+	if err != nil {
+		return false, nil, err
+	}
+	if isAllowedToProceed {
 		// If another request from this transaction has already locked this key with
 		// sufficient locking strength then there's no conflict; we can proceed.
 		return false, nil, nil
@@ -859,7 +866,7 @@ func (g *lockTableGuardImpl) takeToResolveUnreplicated() []roachpb.LockUpdate {
 // etc.
 //
 // ACQUIRES: g.mu.
-func (g *lockTableGuardImpl) resumeScan(notify bool) {
+func (g *lockTableGuardImpl) resumeScan(notify bool) error {
 	spans := g.spans.GetSpans(g.curStrength())
 	var span *roachpb.Span
 	resumingInSameSpan := false
@@ -908,9 +915,12 @@ func (g *lockTableGuardImpl) resumeScan(notify bool) {
 				// Else, past the lock where it stopped waiting. We may not
 				// encounter that lock since it may have been garbage collected.
 			}
-			conflicts := l.scanAndMaybeEnqueue(g, notify)
+			conflicts, err := l.scanAndMaybeEnqueue(g, notify)
+			if err != nil {
+				return err
+			}
 			if conflicts {
-				return
+				return nil
 			}
 		}
 		resumingInSameSpan = false
@@ -951,6 +961,7 @@ func (g *lockTableGuardImpl) resumeScan(notify bool) {
 	if notify {
 		g.notify()
 	}
+	return nil
 }
 
 // queuedGuard is used to wrap waiting locking requests in the keyLocks struct.
@@ -2327,17 +2338,21 @@ func (kl *keyLocks) lockAcquiredOrDiscovered(tl *txnLock) {
 // need resolution.
 //
 // REQUIRES: kl.mu to be locked.
-func (kl *keyLocks) scanAndMaybeEnqueue(g *lockTableGuardImpl, notify bool) (wait bool) {
+func (kl *keyLocks) scanAndMaybeEnqueue(g *lockTableGuardImpl, notify bool) (wait bool, _ error) {
 	kl.mu.Lock()
 	defer kl.mu.Unlock()
 	if kl.isEmptyLock() {
-		return false /* wait */
+		return false /* wait */, nil
 	}
 
 	// It is possible that the lock is already held by this request's
 	// transaction, and it is held with a lock strength good enough for it.
-	if kl.alreadyHoldsLockAndIsAllowedToProceed(g, g.curStrength()) {
-		return false /* wait */
+	isAllowedToProceed, err := kl.alreadyHoldsLockAndIsAllowedToProceed(g, g.curStrength())
+	if err != nil {
+		return false, err
+	}
+	if isAllowedToProceed {
+		return false /* wait */, nil
 	}
 
 	if g.curStrength() == lock.None {
@@ -2345,14 +2360,17 @@ func (kl *keyLocks) scanAndMaybeEnqueue(g *lockTableGuardImpl, notify bool) (wai
 		if conflicts {
 			ws := kl.constructWaitingState(g)
 			g.startWaitingWithWaitingState(ws, notify)
-			return true /* wait */
+			return true /* wait */, nil
 		}
-		return false /* wait */
+		return false /* wait */, nil
 	}
 
 	// We're purely dealing with locking requests from here on out.
 
-	maxQueueLengthExceeded := kl.enqueueLockingRequest(g)
+	maxQueueLengthExceeded, err := kl.enqueueLockingRequest(g)
+	if err != nil {
+		return false, err
+	}
 	if maxQueueLengthExceeded {
 		// NB: Requests that encounter a lock wait-queue that is longer than
 		// what they're willing to wait for are rejected by the lock table
@@ -2362,7 +2380,7 @@ func (kl *keyLocks) scanAndMaybeEnqueue(g *lockTableGuardImpl, notify bool) (wai
 		g.startWaitingWithWaitingState(ws, notify)
 		// Return true, not because we want to wait, but because we want
 		// this request to be rejected in the lock table waiter.
-		return true /* wait */
+		return true /* wait */, nil
 	}
 
 	if kl.shouldRequestActivelyWait(g) {
@@ -2383,14 +2401,14 @@ func (kl *keyLocks) scanAndMaybeEnqueue(g *lockTableGuardImpl, notify bool) (wai
 		// need to call informActiveWaiters. Note that informActiveWaiters elides
 		// updates if they're not meaningful, so we can get away with being less
 		// precise in handling the more general case at this level.
-		return true /* wait */
+		return true /* wait */, nil
 	}
 
 	kl.claimBeforeProceeding(g)
 	// Inform any active waiters that (may) need to be made aware that this
 	// request acquired a claim.
 	kl.informActiveWaiters()
-	return false /* wait */
+	return false /* wait */, nil
 }
 
 // constructWaitingState constructs the waiting state the supplied request
@@ -2424,28 +2442,32 @@ func (kl *keyLocks) constructWaitingState(g *lockTableGuardImpl) waitingState {
 // REQUIRES: kl.mu to be locked.
 func (kl *keyLocks) alreadyHoldsLockAndIsAllowedToProceed(
 	g *lockTableGuardImpl, str lock.Strength,
-) bool {
+) (bool, error) {
 	if !kl.isLocked() {
-		return false // no one holds the lock
+		return false, nil // no one holds the lock
 	}
 	if g.txn == nil {
-		return false // non-transactional requests do not hold locks
+		return false, nil // non-transactional requests do not hold locks
 	}
 	e, found := kl.heldBy[g.txn.ID]
 	if !found {
-		return false
+		return false, nil
 	}
 	tl := e.Value
 	heldMode := tl.getLockMode()
+	err := kl.maybeDisallowLockPromotion(heldMode.Strength, str)
+	if err != nil {
+		return false, err
+	}
 	// Check if the lock is already held by the guard's transaction with an equal
 	// or higher lock strength. If it is, we're good to go. Otherwise, the request
 	// is trying to promote a lock it previously acquired. In such cases, the
 	// existence of a lock with weaker strength doesn't do much for this request.
-	// It's no different than the case where its trying to acquire a fresh lock.
+	// It's no different from the case where it's trying to acquire a fresh lock.
 	return str <= heldMode.Strength ||
 		// TODO(arul): We want to allow requests that are writing to keys that they
 		// hold exclusive locks on to "jump ahead" of any potential waiters. This
-		// prevents deadlocks. The logic here is a bandaid until we implement a
+		// prevents deadlocks. The logic here is a band-aid until we implement a
 		// solution for the general case of arbitrary lock upgrades (e.g. shared ->
 		// exclusive, etc.). We'll do so by prioritizing requests from transaction's
 		// that hold locks over transactions that don't when storing them in the
@@ -2453,7 +2475,27 @@ func (kl *keyLocks) alreadyHoldsLockAndIsAllowedToProceed(
 		// queuedLockingRequests just based on sequence numbers alone, we'll instead
 		// use (belongsToALockHolderTxn, sequence number) to construct the sort
 		// order.
-		(str == lock.Intent && heldMode.Strength == lock.Exclusive)
+		(str == lock.Intent && heldMode.Strength == lock.Exclusive), nil
+}
+
+// maybeDisallowLockPromotion checks if a lock is being promoted from
+// lock.Shared to lock.Intent/lock.Exclusive, and returns an error if that's the
+// case. See: https://github.com/cockroachdb/cockroach/issues/110435.
+//
+// REQUIRES: kl.mu to be locked.
+//
+// TODO(arul): Once we handle lock promotion correctly, and this function goes
+// away, a lot of the error paths where this error is bubbled up to can go away.
+func (kl *keyLocks) maybeDisallowLockPromotion(
+	held lock.Strength, reAcquisitionStr lock.Strength,
+) error {
+	if held == lock.Shared && reAcquisitionStr > held {
+		return errors.UnimplementedErrorf(
+			errors.IssueLink{IssueURL: "https://github.com/cockroachdb/cockroach/issues/110435"},
+			"lock promotion from %s to %s is not allowed", held, reAcquisitionStr,
+		)
+	}
+	return nil
 }
 
 // conflictsWithLockHolders returns true if the request, referenced by the
@@ -2606,7 +2648,9 @@ func (kl *keyLocks) maybeEnqueueNonLockingReadRequest(g *lockTableGuardImpl) (co
 // this case is returned to the caller.
 //
 // REQUIRES: kl.mu to be locked.
-func (kl *keyLocks) enqueueLockingRequest(g *lockTableGuardImpl) (maxQueueLengthExceeded bool) {
+func (kl *keyLocks) enqueueLockingRequest(
+	g *lockTableGuardImpl,
+) (maxQueueLengthExceeded bool, _ error) {
 	assert(g.curStrength() != lock.None, "should only be called with a locking request")
 	g.mu.Lock()
 	defer g.mu.Unlock()
@@ -2624,7 +2668,7 @@ func (kl *keyLocks) enqueueLockingRequest(g *lockTableGuardImpl) (maxQueueLength
 				// it may be a candidate for becoming the distinguished waiter (if one
 				// doesn't exist already).
 				kl.maybeMakeDistinguishedWaiter(g)
-				return false /* maxQueueLengthExceeded */
+				return false /* maxQueueLengthExceeded */, nil
 			}
 		}
 		panic("lock table bug")
@@ -2638,7 +2682,7 @@ func (kl *keyLocks) enqueueLockingRequest(g *lockTableGuardImpl) (maxQueueLength
 		// queue and rejecting the tail of the queue above the max length. That
 		// would be more fair, but more complicated, and we expect that the
 		// common case is that this waiter will be at the end of the queue.
-		return true /* maxQueueLengthExceeded */
+		return true /* maxQueueLengthExceeded */, nil
 	}
 	qg := &queuedGuard{
 		guard:  g,
@@ -2648,22 +2692,31 @@ func (kl *keyLocks) enqueueLockingRequest(g *lockTableGuardImpl) (maxQueueLength
 	// The request isn't in the queue. Add it in the correct position, based on
 	// its sequence number.
 	var e *list.Element[*queuedGuard]
-	for e = kl.queuedLockingRequests.Back(); e != nil; e = e.Prev() {
+	for e = kl.queuedLockingRequests.Front(); e != nil; e = e.Next() {
 		qqg := e.Value
-		if qqg.guard.seqNum < qg.guard.seqNum {
+		if qqg.guard.seqNum > qg.guard.seqNum {
 			break
 		}
+		if qg.guard.txn != nil && qqg.guard.isSameTxn(qg.guard.txnMeta()) {
+			// There's another request, from our transaction, that's waiting to acquire
+			// a lock on this key before us. As per its sequence number, it'll get a
+			// chance to do so before we do -- assuming it'll succeed, check if we've
+			// got ourselves into a lock promotion case that's not allowed.
+			if err := kl.maybeDisallowLockPromotion(qqg.mode.Strength, qg.mode.Strength); err != nil {
+				return false, err
+			}
+		}
 	}
 	if e == nil {
-		kl.queuedLockingRequests.PushFront(qg)
+		kl.queuedLockingRequests.PushBack(qg)
 	} else {
-		kl.queuedLockingRequests.InsertAfter(qg, e)
+		kl.queuedLockingRequests.InsertBefore(qg, e)
 	}
 	// This request may be a candidate to become a distinguished waiter if one
 	// doesn't exist yet; try making it such.
 	kl.maybeMakeDistinguishedWaiter(g)
 	g.maybeAddToLocksMap(kl, g.curStrength())
-	return false /* maxQueueLengthExceeded */
+	return false /* maxQueueLengthExceeded */, nil
 }
 
 // maybeMakeDistinguishedWaiter designates the supplied request as the
@@ -3613,7 +3666,7 @@ func (t *lockTableImpl) ScanOptimistic(req Request) lockTableGuard {
 }
 
 // ScanAndEnqueue implements the lockTable interface.
-func (t *lockTableImpl) ScanAndEnqueue(req Request, guard lockTableGuard) lockTableGuard {
+func (t *lockTableImpl) ScanAndEnqueue(req Request, guard lockTableGuard) (lockTableGuard, *Error) {
 	// NOTE: there is no need to synchronize with enabledMu here. ScanAndEnqueue
 	// scans the lockTable and enters any conflicting lock wait-queues, but a
 	// disabled lockTable will be empty. If the scan's btree snapshot races with
@@ -3642,17 +3695,20 @@ func (t *lockTableImpl) ScanAndEnqueue(req Request, guard lockTableGuard) lockTa
 		// snapshot but does not scan the lock table when sequencing. Instead, it
 		// calls into IsKeyLockedByConflictingTxn before adding keys to its result
 		// set to determine which keys it should skip.
-		return g
+		return g, nil
 	}
 
-	g.resumeScan(true /* notify */)
+	err := g.resumeScan(true /* notify */)
+	if err != nil {
+		return nil, kvpb.NewError(err)
+	}
 	if g.notRemovableLock != nil {
 		// Either waiting at the notRemovableLock, or elsewhere. Either way we are
 		// making forward progress, which ensures liveness.
 		g.notRemovableLock.decrementNotRemovable()
 		g.notRemovableLock = nil
 	}
-	return g
+	return g, nil
 }
 
 func (t *lockTableImpl) newGuardForReq(req Request) *lockTableGuardImpl {
diff --git a/pkg/kv/kvserver/concurrency/lock_table_test.go b/pkg/kv/kvserver/concurrency/lock_table_test.go
index 27de15296ad5..8bf9fee51cfe 100644
--- a/pkg/kv/kvserver/concurrency/lock_table_test.go
+++ b/pkg/kv/kvserver/concurrency/lock_table_test.go
@@ -347,7 +347,11 @@ func TestLockTableBasic(t *testing.T) {
 					d.Fatalf(t, "unknown request: %s", reqName)
 				}
 				g := guardsByReqName[reqName]
-				g = lt.ScanAndEnqueue(req, g)
+				var err *Error
+				g, err = lt.ScanAndEnqueue(req, g)
+				if err != nil {
+					return err.String()
+				}
 				guardsByReqName[reqName] = g
 				return fmt.Sprintf("start-waiting: %t", g.ShouldWait())
 
@@ -553,7 +557,10 @@ func TestLockTableBasic(t *testing.T) {
 				default:
 					str = "old: "
 				}
-				state := g.CurState()
+				state, err := g.CurState()
+				if err != nil {
+					return err.Error()
+				}
 				var typeStr string
 				switch state.kind {
 				case waitForDistinguished:
@@ -858,7 +865,8 @@ func TestLockTableMaxLocks(t *testing.T) {
 			LockSpans:  lockSpans,
 		}
 		reqs = append(reqs, req)
-		ltg := lt.ScanAndEnqueue(req, nil)
+		ltg, err := lt.ScanAndEnqueue(req, nil)
+		require.Nil(t, err)
 		require.Nil(t, ltg.ResolveBeforeScanning())
 		require.False(t, ltg.ShouldWait())
 		guards = append(guards, ltg)
@@ -885,7 +893,9 @@ func TestLockTableMaxLocks(t *testing.T) {
 	require.Equal(t, int64(10), lt.lockCountForTesting())
 	// Two guards do ScanAndEnqueue.
 	for i := 2; i < 4; i++ {
-		guards[i] = lt.ScanAndEnqueue(reqs[i], guards[i])
+		var err *Error
+		guards[i], err = lt.ScanAndEnqueue(reqs[i], guards[i])
+		require.Nil(t, err)
 		require.True(t, guards[i].ShouldWait())
 	}
 	require.Equal(t, int64(10), lt.lockCountForTesting())
@@ -991,7 +1001,8 @@ func TestLockTableMaxLocksWithMultipleNotRemovableRefs(t *testing.T) {
 			LatchSpans: latchSpans,
 			LockSpans:  lockSpans,
 		}
-		ltg := lt.ScanAndEnqueue(req, nil)
+		ltg, err := lt.ScanAndEnqueue(req, nil)
+		require.Nil(t, err)
 		require.Nil(t, ltg.ResolveBeforeScanning())
 		require.False(t, ltg.ShouldWait())
 		guards = append(guards, ltg)
@@ -1079,7 +1090,11 @@ func doWork(ctx context.Context, item *workItem, e *workloadExecutor) error {
 			if err != nil {
 				return err
 			}
-			g = e.lt.ScanAndEnqueue(*item.request, g)
+			var kvErr *Error
+			g, kvErr = e.lt.ScanAndEnqueue(*item.request, g)
+			if kvErr != nil {
+				return kvErr.GoError()
+			}
 			if !g.ShouldWait() {
 				break
 			}
@@ -1092,7 +1107,10 @@ func doWork(ctx context.Context, item *workItem, e *workloadExecutor) error {
 				case <-ctx.Done():
 					return ctx.Err()
 				}
-				state := g.CurState()
+				state, err := g.CurState()
+				if err != nil {
+					return err
+				}
 				switch state.kind {
 				case doneWaiting:
 					if !lastID.Equal(uuid.UUID{}) && item.request.Txn != nil {
@@ -1638,7 +1656,12 @@ func doBenchWork(item *benchWorkItem, env benchEnv, doneCh chan<- error) {
 			doneCh <- err
 			return
 		}
-		g = env.lt.ScanAndEnqueue(item.Request, g)
+		var kvErr *Error
+		g, kvErr = env.lt.ScanAndEnqueue(item.Request, g)
+		if kvErr != nil {
+			doneCh <- kvErr.GoError()
+			return
+		}
 		atomic.AddUint64(env.numScanCalls, 1)
 		if !g.ShouldWait() {
 			break
@@ -1650,7 +1673,11 @@ func doBenchWork(item *benchWorkItem, env benchEnv, doneCh chan<- error) {
 		env.lm.Release(lg)
 		for {
 			<-g.NewStateChan()
-			state := g.CurState()
+			state, err := g.CurState()
+			if err != nil {
+				doneCh <- err
+				return
+			}
 			if state.kind == doneWaiting {
 				break
 			}
diff --git a/pkg/kv/kvserver/concurrency/lock_table_waiter.go b/pkg/kv/kvserver/concurrency/lock_table_waiter.go
index 16f9854f358c..bba7c71feb5d 100644
--- a/pkg/kv/kvserver/concurrency/lock_table_waiter.go
+++ b/pkg/kv/kvserver/concurrency/lock_table_waiter.go
@@ -161,7 +161,10 @@ func (w *lockTableWaiterImpl) WaitOn(
 		// about another contending transaction on newStateC.
 		case <-newStateC:
 			timerC = nil
-			state := guard.CurState()
+			state, err := guard.CurState()
+			if err != nil {
+				return kvpb.NewError(err)
+			}
 			log.VEventf(ctx, 3, "lock wait-queue event: %s", state)
 			tracer.notify(ctx, state)
 			switch state.kind {
diff --git a/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go b/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go
index 2ed3c83f1bb4..5e385053ebff 100644
--- a/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go
+++ b/pkg/kv/kvserver/concurrency/lock_table_waiter_test.go
@@ -74,12 +74,12 @@ var _ lockTableGuard = &mockLockTableGuard{}
 // mockLockTableGuard implements the lockTableGuard interface.
 func (g *mockLockTableGuard) ShouldWait() bool            { return true }
 func (g *mockLockTableGuard) NewStateChan() chan struct{} { return g.signal }
-func (g *mockLockTableGuard) CurState() waitingState {
+func (g *mockLockTableGuard) CurState() (waitingState, error) {
 	s := g.state
 	if g.stateObserved != nil {
 		g.stateObserved <- struct{}{}
 	}
-	return s
+	return s, nil
 }
 func (g *mockLockTableGuard) ResolveBeforeScanning() []roachpb.LockUpdate {
 	return g.toResolve
diff --git a/pkg/kv/kvserver/concurrency/testdata/lock_table/shared_locks b/pkg/kv/kvserver/concurrency/testdata/lock_table/shared_locks
index 0a808d25c4de..104023d77983 100644
--- a/pkg/kv/kvserver/concurrency/testdata/lock_table/shared_locks
+++ b/pkg/kv/kvserver/concurrency/testdata/lock_table/shared_locks
@@ -86,14 +86,14 @@ num=1
            txn: 00000000-0000-0000-0000-000000000002 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
 
 
-new-request r=req5 txn=txn2 ts=10 spans=exclusive@a
+new-request r=req5 txn=txn4 ts=10 spans=exclusive@a
 ----
 
 scan r=req5
 ----
 start-waiting: true
 
-new-request r=req6 txn=txn2 ts=10 spans=intent@a
+new-request r=req6 txn=txn4 ts=10 spans=intent@a
 ----
 
 scan r=req6
@@ -107,8 +107,8 @@ num=1
   holders: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
            txn: 00000000-0000-0000-0000-000000000002 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
    queued locking requests:
-    active: true req: 5, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000002
-    active: true req: 6, strength: Intent, txn: 00000000-0000-0000-0000-000000000002
+    active: true req: 5, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000004
+    active: true req: 6, strength: Intent, txn: 00000000-0000-0000-0000-000000000004
    distinguished req: 5
 
 # ------------------------------------------------------------------------------
@@ -131,8 +131,8 @@ num=1
   holders: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
            txn: 00000000-0000-0000-0000-000000000002 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
    queued locking requests:
-    active: true req: 5, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000002
-    active: true req: 6, strength: Intent, txn: 00000000-0000-0000-0000-000000000002
+    active: true req: 5, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000004
+    active: true req: 6, strength: Intent, txn: 00000000-0000-0000-0000-000000000004
     active: true req: 7, strength: Shared, txn: 00000000-0000-0000-0000-000000000003
    distinguished req: 5
 
@@ -153,8 +153,8 @@ num=1
   holders: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
            txn: 00000000-0000-0000-0000-000000000002 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
    queued locking requests:
-    active: true req: 5, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000002
-    active: true req: 6, strength: Intent, txn: 00000000-0000-0000-0000-000000000002
+    active: true req: 5, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000004
+    active: true req: 6, strength: Intent, txn: 00000000-0000-0000-0000-000000000004
     active: true req: 7, strength: Shared, txn: 00000000-0000-0000-0000-000000000003
    distinguished req: 5
 
@@ -934,6 +934,142 @@ num=1
     active: true req: 46, strength: Shared, txn: 00000000-0000-0000-0000-000000000003
    distinguished req: 46
 
+# ------------------------------------------------------------------------------
+# Test for lock promotion. When a lock is held with strength Shared, we do not
+# allow the holder to promote it to Exclusive or write to the key. The same
+# applies if the lock isn't held, but there's a request from our transaction
+# trying to acquire a Shared lock (that's waiting in front of us).
+# ------------------------------------------------------------------------------
+
+clear
+----
+num=0
+
+new-request r=req47 txn=txn1 ts=10 spans=shared@b
+----
+
+scan r=req47
+----
+start-waiting: false
+
+acquire k=b r=req47 strength=shared durability=u
+----
+num=1
+ lock: "b"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
+
+new-request r=req48 txn=txn1 ts=10 spans=exclusive@b
+----
+
+scan r=req48
+----
+lock promotion from Shared to Exclusive is not allowed
+
+new-request r=req49 txn=txn1 ts=10 spans=intent@b
+----
+
+scan r=req49
+----
+lock promotion from Shared to Intent is not allowed
+
+new-request r=req50 txn=txn2 ts=10 spans=exclusive@b
+----
+
+scan r=req50
+----
+start-waiting: true
+
+new-request r=req51 txn=txn1 ts=10 spans=exclusive@a
+----
+
+scan r=req51
+----
+start-waiting: false
+
+acquire r=req51 k=a durability=u strength=exclusive
+----
+num=2
+ lock: "a"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, ts: 10.000000000,0, info: unrepl [(str: Exclusive seq: 0)]
+ lock: "b"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
+   queued locking requests:
+    active: true req: 50, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000002
+   distinguished req: 50
+
+new-request r=req52 txn=txn3 ts=10 spans=exclusive@a+exclusive@b
+----
+
+scan r=req52
+----
+start-waiting: true
+
+new-request r=req53 txn=txn3 ts=10 spans=shared@b
+----
+
+scan r=req53
+----
+start-waiting: true
+
+print
+----
+num=2
+ lock: "a"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, ts: 10.000000000,0, info: unrepl [(str: Exclusive seq: 0)]
+   queued locking requests:
+    active: true req: 52, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000003
+   distinguished req: 52
+ lock: "b"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
+   queued locking requests:
+    active: true req: 50, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000002
+    active: true req: 53, strength: Shared, txn: 00000000-0000-0000-0000-000000000003
+   distinguished req: 50
+
+new-request r=req54 txn=txn3 ts=10 spans=exclusive@b
+----
+
+scan r=req54
+----
+lock promotion from Shared to Exclusive is not allowed
+
+new-request r=req55 txn=txn3 ts=10 spans=intent@b
+----
+
+scan r=req55
+----
+lock promotion from Shared to Intent is not allowed
+
+release txn=txn1 span=a
+----
+num=2
+ lock: "a"
+   queued locking requests:
+    active: false req: 52, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000003
+ lock: "b"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
+   queued locking requests:
+    active: true req: 50, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000002
+    active: true req: 53, strength: Shared, txn: 00000000-0000-0000-0000-000000000003
+   distinguished req: 50
+
+scan r=req52
+----
+start-waiting: true
+
+print
+----
+num=2
+ lock: "a"
+   queued locking requests:
+    active: false req: 52, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000003
+ lock: "b"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, info: unrepl [(str: Shared seq: 0)]
+   queued locking requests:
+    active: true req: 50, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000002
+    active: true req: 52, strength: Exclusive, txn: 00000000-0000-0000-0000-000000000003
+    active: true req: 53, strength: Shared, txn: 00000000-0000-0000-0000-000000000003
+   distinguished req: 50
 
 # TODO(arul): (non-exhaustive list) of shared lock state transitions that aren't
 # currently supported (and we need to add support for):
@@ -947,8 +1083,6 @@ num=1
 # reservation). Ditto for a partial break, where the exclusive locking request
 # inserts itself in the middle of the queue.
 #
-# 3. Lock promotion cases where the lock is held with strength shared, and
-# there's another request (from the same txn) with lock strength exclusive
-# that's trying to acquire the lock further back in the wait queue. We'll be
-# disallowing these in the short term, so this case might not be tested for a
-# while.
+# 3. Lock promotion -- add a test where we discover that we're trying to promote
+# a lock from shared -> exclusive when resuming a scan from the lock table
+# waiter. An error should be returned to the client.
diff --git a/pkg/kv/kvserver/concurrency/testdata/lock_table/skip_locked b/pkg/kv/kvserver/concurrency/testdata/lock_table/skip_locked
index 034d045eef6c..9bec32190bf9 100644
--- a/pkg/kv/kvserver/concurrency/testdata/lock_table/skip_locked
+++ b/pkg/kv/kvserver/concurrency/testdata/lock_table/skip_locked
@@ -356,11 +356,11 @@ locked: true, holder: 00000000-0000-0000-0000-000000000001
 
 is-key-locked-by-conflicting-txn r=req8 k=h strength=exclusive
 ----
-locked: false
+lock promotion from Shared to Exclusive is not allowed
 
 is-key-locked-by-conflicting-txn r=req8 k=i strength=exclusive
 ----
-locked: true, holder: <nil>
+lock promotion from Shared to Exclusive is not allowed
 
 dequeue r=req8
 ----