kvserver: add per-operation lock reliability settings

Preserving unreplicated locks during split, merge, and lease transfers have
different trade offs. For instance, during a split all lock updates
are done in memory without any new replicated writes, whereas for
merge and lease transfers requiring replicating locks through raft.

Here, we put the different operations under different settings since
we may want to ship different defaults for the different operations.

Epic: none
Release note: None

Author: stevendanna

Diff for: pkg/kv/kvserver/client_replica_test.go

@@ -5781,7 +5781,7 @@ func TestLeaseTransferReplicatesLocks(t *testing.T) {
 	// txn2 is never unblocked (from the perspective of the client).
 	ctx := context.Background()
 	st := cluster.MakeClusterSettings()
-	concurrency.UnreplicatedLockReliability.Override(ctx, &st.SV, true)
+	concurrency.UnreplicatedLockReliabilityLeaseTransfer.Override(ctx, &st.SV, true)
 	tc := testcluster.StartTestCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Settings: st,
@@ -5891,7 +5891,7 @@ func TestMergeReplicatesLocks(t *testing.T) {
 		ctx = context.Background()
 		st  = cluster.MakeClusterSettings()
 	)
-	concurrency.UnreplicatedLockReliability.Override(ctx, &st.SV, true)
+	concurrency.UnreplicatedLockReliabilityMerge.Override(ctx, &st.SV, true)
 
 	for _, b := range []bool{true, false} {
 		name := "lhs-lock"

Diff for: pkg/kv/kvserver/concurrency/concurrency_manager.go

@@ -114,13 +114,31 @@ var BatchPushedLockResolution = settings.RegisterBoolSetting(
 	true,
 )
 
-// UnreplicatedLockReliability controls whether the replica will attempt
-// to keep unreplicated locks during node operations such as split.
-var UnreplicatedLockReliability = settings.RegisterBoolSetting(
+// UnreplicatedLockReliabilitySplit controls whether the replica will attempt
+// to keep unreplicated locks during range split operations.
+var UnreplicatedLockReliabilitySplit = settings.RegisterBoolSetting(
 	settings.SystemOnly,
-	"kv.lock_table.unreplicated_lock_reliability.enabled",
-	"whether the replica should attempt to keep unreplicated locks during various node operations",
-	metamorphic.ConstantWithTestBool("kv.lock_table.unreplicated_lock_reliability.enabled", true),
+	"kv.lock_table.unreplicated_lock_reliability.split.enabled",
+	"whether the replica should attempt to keep unreplicated locks during range splits",
+	metamorphic.ConstantWithTestBool("kv.lock_table.unreplicated_lock_reliability.split.enabled", true),
+)
+
+// UnreplicatedLockReliabilityLeaseTransfer controls whether the replica will attempt
+// to keep unreplicated locks during lease transfer operations.
+var UnreplicatedLockReliabilityLeaseTransfer = settings.RegisterBoolSetting(
+	settings.SystemOnly,
+	"kv.lock_table.unreplicated_lock_reliability.lease_transfer.enabled",
+	"whether the replica should attempt to keep unreplicated locks during lease transfers",
+	metamorphic.ConstantWithTestBool("kv.lock_table.unreplicated_lock_reliability.lease_transfer.enabled", true),
+)
+
+// UnreplicatedLockReliabilityMerge controls whether the replica will
+// attempt to keep unreplicated locks during range merge operations.
+var UnreplicatedLockReliabilityMerge = settings.RegisterBoolSetting(
+	settings.SystemOnly,
+	"kv.lock_table.unreplicated_lock_reliability.merge.enabled",
+	"whether the replica should attempt to keep unreplicated locks during range merges",
+	metamorphic.ConstantWithTestBool("kv.lock_table.unreplicated_lock_reliability.merge.enabled", true),
 )
 
 // managerImpl implements the Manager interface.
@@ -590,7 +608,7 @@ var allKeysSpan = roachpb.Span{Key: keys.MinKey, EndKey: keys.MaxKey}
 
 // OnRangeLeaseTransferEval implements the RangeStateListener interface.
 func (m *managerImpl) OnRangeLeaseTransferEval() []*roachpb.LockAcquisition {
-	if !UnreplicatedLockReliability.Get(&m.st.SV) {
+	if !UnreplicatedLockReliabilityLeaseTransfer.Get(&m.st.SV) {
 		return nil
 	}
 
@@ -606,7 +624,7 @@ func (m *managerImpl) OnRangeLeaseTransferEval() []*roachpb.LockAcquisition {
 // during evalutation of Subsume. The returned LockAcquisition structs represent
 // held locks that we may want to flush to disk as replicated.
 func (m *managerImpl) OnRangeSubsumeEval() []*roachpb.LockAcquisition {
-	if !UnreplicatedLockReliability.Get(&m.st.SV) {
+	if !UnreplicatedLockReliabilityMerge.Get(&m.st.SV) {
 		return nil
 	}
 
@@ -636,7 +654,7 @@ func (m *managerImpl) OnRangeLeaseUpdated(seq roachpb.LeaseSequence, isLeasehold
 // LHS replica of a split and should be passed the new RHS start key (LHS
 // EndKey).
 func (m *managerImpl) OnRangeSplit(rhsStartKey roachpb.Key) []roachpb.LockAcquisition {
-	if UnreplicatedLockReliability.Get(&m.st.SV) {
+	if UnreplicatedLockReliabilitySplit.Get(&m.st.SV) {
 		lockToMove := m.lt.ClearGE(rhsStartKey)
 		m.twq.ClearGE(rhsStartKey)
 		return lockToMove

Diff for: pkg/kv/kvserver/concurrency/concurrency_manager_test.go

@@ -735,7 +735,9 @@ func newClusterWithSettings(st *clustersettings.Settings) *cluster {
 	// Set the latch manager's long latch threshold to infinity to disable
 	// logging, which could cause a test to erroneously fail.
 	spanlatch.LongLatchHoldThreshold.Override(context.Background(), &st.SV, math.MaxInt64)
-	concurrency.UnreplicatedLockReliability.Override(context.Background(), &st.SV, true)
+	concurrency.UnreplicatedLockReliabilitySplit.Override(context.Background(), &st.SV, true)
+	concurrency.UnreplicatedLockReliabilityMerge.Override(context.Background(), &st.SV, true)
+	concurrency.UnreplicatedLockReliabilityLeaseTransfer.Override(context.Background(), &st.SV, true)
 	manual := timeutil.NewManualTime(timeutil.Unix(123, 0))
 	return &cluster{
 		nodeDesc:  &roachpb.NodeDescriptor{NodeID: 1},

Diff for: pkg/kv/kvserver/replica_command.go

@@ -915,7 +915,7 @@ func (r *Replica) AdminMerge(
 		// This must be a single request in a BatchRequest: there are multiple
 		// places that do special logic (needed for safety) that rely on
 		// BatchRequest.IsSingleSubsumeRequest() returning true.
-		shouldPreserveLocks := concurrency.UnreplicatedLockReliability.Get(&r.ClusterSettings().SV)
+		shouldPreserveLocks := concurrency.UnreplicatedLockReliabilityMerge.Get(&r.ClusterSettings().SV)
 		br, pErr := kv.SendWrapped(ctx, r.store.DB().NonTransactionalSender(),
 			&kvpb.SubsumeRequest{
 				RequestHeader: kvpb.RequestHeader{