concurrency: enable joint claims on the request scan path

This patch addresses a TODO that was blocking joint claims on a
request's scan path and adds a test to show joint claims work. In
particular, previously the lock mode of a queued request was hardcoded
to use locking strength `lock.Intent`. We now use the correct lock
mode by snapshotting an immutable copy of the request's lock strength
on to the queuedGuard when inserting a it into a lock's wait queue.

Informs cockroachdb#102275

Release note: None

pkg/kv/kvserver/concurrency/lock_table.go

@@ -919,6 +919,7 @@ func (g *lockTableGuardImpl) resumeScan(notify bool) {
 // and non-transactional requests.
 type queuedGuard struct {
 	guard  *lockTableGuardImpl
+	mode   lock.Mode
 	active bool // protected by lockState.mu
 }
 
@@ -2201,6 +2202,7 @@ func (l *lockState) enqueueLockingRequest(g *lockTableGuardImpl) (maxQueueLength
 	}
 	qg := &queuedGuard{
 		guard:  g,
+		mode:   g.curLockMode(),
 		active: true,
 	}
 	// The request isn't in the queue. Add it in the correct position, based on
@@ -2281,13 +2283,7 @@ func (l *lockState) shouldRequestActivelyWait(g *lockTableGuardImpl) bool {
 			// conflicting waiters; no need to actively wait here.
 			return false
 		}
-		// TODO(arul): Inactive waiters will need to capture the strength at which
-		// they're trying to acquire a lock in their queuedGuard. We can't simply
-		// use the guard's curStrength (or curLockMode) -- inactive waiters may have
-		// mutated these values as they scan. For now, we can just use the intent
-		// lock mode as that's the only lock strength supported by the lock table.
-		waiterLockMode := lock.MakeModeIntent(qqg.guard.ts)
-		if lock.Conflicts(waiterLockMode, g.curLockMode(), &g.lt.settings.SV) {
+		if lock.Conflicts(qqg.mode, g.curLockMode(), &g.lt.settings.SV) {
 			return true
 		}
 	}
@@ -2688,9 +2684,23 @@ func (l *lockState) discoveredLock(
 		g.mu.Unlock()
 
 		if !presentHere {
+			var m lock.Mode
+			switch accessStrength {
+			case lock.None:
+				panic("cannot add non-locking request to the list of queuedWriters")
+			case lock.Shared:
+				m = lock.MakeModeShared()
+			case lock.Exclusive:
+				m = lock.MakeModeExclusive(g.ts, g.txn.IsoLevel)
+			case lock.Intent:
+				m = lock.MakeModeIntent(g.ts)
+			default:
+				panic(fmt.Sprintf("unhandled request strength: %s", g.curStrength()))
+			}
 			// Put self in queue as inactive waiter.
 			qg := &queuedGuard{
 				guard:  g,
+				mode:   m,
 				active: false,
 			}
 			// g is not necessarily first in the queue in the (rare) case (a) above.

pkg/kv/kvserver/concurrency/testdata/lock_table/shared_locks

@@ -101,3 +101,44 @@ num=1
     active: true req: 4, txn: 00000000-0000-0000-0000-000000000002
     active: true req: 5, txn: 00000000-0000-0000-0000-000000000002
    distinguished req: 3
+
+# ------------------------------------------------------------------------------
+# Ensure scanning requests are able to acquire joint claims on the initial
+# scan path.
+# ------------------------------------------------------------------------------
+
+clear
+----
+num=0
+
+new-request r=req7 txn=txn1 ts=10 spans=shared@a
+----
+
+acquire r=req1 k=a durability=u strength=shared
+----
+num=1
+ lock: "a"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, ts: 10.000000000,0, info: unrepl [(str: Shared seq: 0)]
+
+new-request r=req8 txn=txn2 ts=10 spans=shared@a
+----
+
+scan r=req8
+----
+start-waiting: false
+
+new-request r=req9 txn=txn2 ts=10 spans=shared@a
+----
+
+scan r=req9
+----
+start-waiting: false
+
+print
+----
+num=1
+ lock: "a"
+  holder: txn: 00000000-0000-0000-0000-000000000001 epoch: 0, iso: Serializable, ts: 10.000000000,0, info: unrepl [(str: Shared seq: 0)]
+   queued writers:
+    active: false req: 6, txn: 00000000-0000-0000-0000-000000000002
+    active: false req: 7, txn: 00000000-0000-0000-0000-000000000002