kvserver: add basic FlushLockTable request

This adds a new request FlushLockTableRequest. When issued, any
unreplicated locks in the request header's span will be written as
replicated locks to disk and removed from the in-memory lock table.

This is an unsplittable request that can't span range boundaries.

The goal of this request is two fold:

1) By modelling this as a request, we can easily issue flush
   operations during KVNemesis at random. This will help us uncover
   problems with our current flush semantics.

2) We may use this flush request from a new queue that will
   periodically ensure that the size of the lock table is
   under control.

Epic: CRDB-42764

Release note: None

Author: stevendanna

docs/generated/metrics/metrics.html

@@ -2980,6 +2980,18 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: distsender.rpc.flushlocktable.sent
+      exported_name: distsender_rpc_flushlocktable_sent
+      description: |-
+        Number of FlushLockTable requests processed.
+
+        This counts the requests in batches handed to DistSender, not the RPCs
+        sent to individual Ranges as a result.
+      y_axis_label: RPCs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: distsender.rpc.gc.sent
       exported_name: distsender_rpc_gc_sent
       description: |-
@@ -15568,6 +15580,14 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: rpc.method.flushlocktable.recv
+      exported_name: rpc_method_flushlocktable_recv
+      description: Number of FlushLockTable requests processed
+      y_axis_label: RPCs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: rpc.method.gc.recv
       exported_name: rpc_method_gc_recv
       description: Number of GC requests processed

docs/generated/metrics/metrics.yaml

@@ -985,6 +985,9 @@ func (*BarrierRequest) Method() Method { return Barrier }
 // Method implements the Request interface.
 func (*IsSpanEmptyRequest) Method() Method { return IsSpanEmpty }
 
+// Method implements the Request interface.
+func (*FlushLockTableRequest) Method() Method { return FlushLockTable }
+
 // ShallowCopy implements the Request interface.
 func (gr *GetRequest) ShallowCopy() Request {
 	shallowCopy := *gr
@@ -1273,6 +1276,12 @@ func (r *IsSpanEmptyRequest) ShallowCopy() Request {
 	return &shallowCopy
 }
 
+// ShallowCopy implements the Request interface.
+func (r *FlushLockTableRequest) ShallowCopy() Request {
+	shallowCopy := *r
+	return &shallowCopy
+}
+
 // ShallowCopy implements the Response interface.
 func (gr *GetResponse) ShallowCopy() Response {
 	shallowCopy := *gr
@@ -1559,6 +1568,12 @@ func (r *IsSpanEmptyResponse) ShallowCopy() Response {
 	return &shallowCopy
 }
 
+// ShallowCopy implements the Response interface.
+func (r *FlushLockTableResponse) ShallowCopy() Response {
+	shallowCopy := *r
+	return &shallowCopy
+}
+
 // NewLockingGet returns a Request initialized to get the value at key. A lock
 // corresponding to the supplied lock strength and durability is acquired on the
 // key, if it exists.
@@ -2087,6 +2102,9 @@ func (r *BarrierRequest) flags() flag {
 	return flags
 }
 func (*IsSpanEmptyRequest) flags() flag { return isRead | isRange }
+func (*FlushLockTableRequest) flags() flag {
+	return isWrite | isRange | isAlone | isUnsplittable
+}
 
 // IsParallelCommit returns whether the EndTxn request is attempting to perform
 // a parallel commit. See txn_interceptor_committer.go for a discussion about

pkg/kv/kvpb/api.go

@@ -2501,6 +2501,22 @@ message BarrierResponse {
   RangeDescriptor range_desc = 4 [(gogoproto.nullable) = false];
 }
 
+// FlushLockTableRequest instructs the replica, if it is the leaseholder, to
+// flush any in-memory unreplicated locks to the replicated lock table contained
+// in the request header's span.
+message FlushLockTableRequest {
+  RequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+}
+
+// FlushLockTableResponse is the response for a FlushLockTable operation.
+message FlushLockTableResponse {
+  ResponseHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+
+  // LocksWritten is the number of locks written to the replicated lock table as
+  // a result of this request.
+  uint64 locks_written = 2;
+}
+
 // A RequestUnion contains exactly one of the requests.
 // The values added here must match those in ResponseUnion.
 //
@@ -2556,6 +2572,7 @@ message RequestUnion {
     IsSpanEmptyRequest is_span_empty = 56;
     LinkExternalSSTableRequest link_external_sstable = 57;
     ExciseRequest excise = 58;
+    FlushLockTableRequest flush_lock_table = 59;
   }
   reserved 8, 15, 23, 25, 26, 27, 31, 34, 49, 52;
 }
@@ -2611,6 +2628,7 @@ message ResponseUnion {
     IsSpanEmptyResponse is_span_empty = 56;
     LinkExternalSSTableResponse link_external_sstable = 57;
     ExciseResponse excise = 58;
+    FlushLockTableResponse flush_lock_table = 59;
   }
   reserved 8, 15, 23, 25, 26, 27, 28, 31, 34, 49, 52;
 }

pkg/kv/kvpb/api.proto

@@ -167,6 +167,11 @@ const (
 	// Excise is a non-MVCC command that destroys all data in a user MVCC key
 	// span. See ExciseRequest for details.
 	Excise
+
+	// FlushLockTable is an operation writes unreplicated locks in the
+	// in-memory lock table to the replicated lock table.
+	FlushLockTable
+
 	// MaxMethod is the maximum method.
 	MaxMethod Method = iota - 1
 	// NumMethods represents the total number of API methods.

pkg/kv/kvpb/batch_generated.go

@@ -13,6 +13,7 @@ go_library(
         "cmd_end_transaction.go",
         "cmd_excise.go",
         "cmd_export.go",
+        "cmd_flush_lock_table.go",
         "cmd_gc.go",
         "cmd_get.go",
         "cmd_heartbeat_txn.go",

pkg/kv/kvpb/method.go

@@ -0,0 +1,69 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package batcheval
+
+import (
+	"context"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/batcheval/result"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/lockspanset"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/storage"
+)
+
+func init() {
+	RegisterReadWriteCommand(kvpb.FlushLockTable, declareKeysFlushLockTable, FlushLockTable)
+}
+
+func declareKeysFlushLockTable(
+	_ ImmutableRangeState,
+	_ *kvpb.Header,
+	req kvpb.Request,
+	latchSpans *spanset.SpanSet,
+	_ *lockspanset.LockSpanSet,
+	_ time.Duration,
+) error {
+	// We declare non-MVCC read-write latches over the entire span we are
+	// exporting. This is similar to the latches LeaseTransfer and Merge will
+	// take. This is perhaps more aggressive than needed.
+	//
+	// TODO(ssd): Consider moving this back to normal MVCC write latches.
+	latchSpans.AddNonMVCC(spanset.SpanReadWrite, req.Header().Span())
+	return nil
+}
+
+// FlushLockTable scans the in-memory lock tbale for unreplicated locks and
+// writes them as replicated locks.
+func FlushLockTable(
+	ctx context.Context, rw storage.ReadWriter, cArgs CommandArgs, response kvpb.Response,
+) (result.Result, error) {
+	args := cArgs.Args.(*kvpb.FlushLockTableRequest)
+	resp := response.(*kvpb.FlushLockTableResponse)
+
+	// TODO(ssd): Allow the caller to limit how many locks we write out.
+	locksToFlush := make([]roachpb.LockAcquisition, 0)
+	cArgs.EvalCtx.GetConcurrencyManager().ExportUnreplicatedLocks(args.Span(), func(l *roachpb.LockAcquisition) {
+		locksToFlush = append(locksToFlush, *l)
+	})
+
+	for i, l := range locksToFlush {
+		locksToFlush[i].Durability = lock.Replicated
+		if err := storage.MVCCAcquireLock(ctx, rw,
+			&l.Txn, l.IgnoredSeqNums, l.Strength, l.Key,
+			cArgs.Stats, 0, 0, true /* allowSequenceNumberRegression */); err != nil {
+			return result.Result{}, err
+		}
+	}
+	resp.LocksWritten = uint64(len(locksToFlush))
+
+	// NOTE: The locks still exist in the in-memory lock table. They are not
+	// cleared until OnLockAcquired is called by (*replica).handleReadWriteLocalEvalResult.
+	return result.WithAcquiredLocks(locksToFlush...), nil
+}

pkg/kv/kvpb/method_string.go

@@ -159,6 +159,20 @@ func TestClientLockTableDataDriven(t *testing.T) {
 					return fmt.Sprintf("error: %s", err.Error())
 				}
 				return ""
+			case "flush-lock-table":
+				startKey := evalCtx.getNamedKey("start", d)
+				endKey := evalCtx.getNamedKey("end", d)
+				b := db.NewBatch()
+				b.AddRawRequest(&kvpb.FlushLockTableRequest{
+					RequestHeader: kvpb.RequestHeader{
+						Key:    startKey,
+						EndKey: endKey,
+					},
+				})
+				if err := db.Run(ctx, b); err != nil {
+					return fmt.Sprintf("error: %s", err.Error())
+				}
+				return ""
 			default:
 				d.Fatalf(t, "unknown command: %s", d.Cmd)
 				return ""