Merge #143124

143124: kvserver: go through DB for GC r=tbg a=tbg

Instead of doing a bespoke thing that hands requests to the replica directly
and bypasses a lot of plumbing, go through the DB. This means GCRequest enters
the Replica like every other request, after passing through the DistSender
stack and being routed through the Node.

This avoids surprises such as discovered in #143122.

Epic: none
Release note: None


Co-authored-by: Tobias Grieger <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/kvserverbase/base.go

@@ -132,14 +132,15 @@ var _ redact.SafeFormatter = CmdIDKey("")
 
 // FilterArgs groups the arguments to a ReplicaCommandFilter.
 type FilterArgs struct {
-	Ctx     context.Context
-	CmdID   CmdIDKey
-	Index   int
-	Sid     roachpb.StoreID
-	Req     kvpb.Request
-	Hdr     kvpb.Header
-	Version roachpb.Version
-	Err     error // only used for TestingPostEvalFilter
+	Ctx          context.Context
+	CmdID        CmdIDKey
+	Index        int
+	Sid          roachpb.StoreID
+	Req          kvpb.Request
+	Hdr          kvpb.Header
+	AdmissionHdr kvpb.AdmissionHeader
+	Version      roachpb.Version
+	Err          error // only used for TestingPostEvalFilter
 }
 
 // ProposalFilterArgs groups the arguments to ReplicaProposalFilter.

pkg/kv/kvserver/mvcc_gc_queue.go

@@ -13,6 +13,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/gc"
@@ -123,16 +124,6 @@ var EnqueueInMvccGCQueueOnSpanConfigUpdateEnabled = settings.RegisterBoolSetting
 	false,
 )
 
-// See https://github.com/cockroachdb/cockroach/pull/143122.
-var mvccGCQueueFullyEnableAC = settings.RegisterBoolSetting(
-	settings.SystemOnly,
-	"kv.mvcc_gc.queue_kv_admission_control.enabled",
-	"when true, MVCC GC queue operations are subject to store admission control. If set to false, "+
-		"since store admission control will be disabled, replication flow control will also be effectively disabled. "+
-		"This setting does not affect CPU admission control.",
-	true,
-)
-
 func largeAbortSpan(ms enginepb.MVCCStats) bool {
 	// Checks if the size of the abort span exceeds the given threshold.
 	// The abort span is not supposed to become that large, but it does
@@ -594,6 +585,11 @@ func (r *replicaGCer) template() kvpb.GCRequest {
 	desc := r.repl.Desc()
 	var template kvpb.GCRequest
 	template.Key = desc.StartKey.AsRawKey()
+	if r.repl.RangeID == 1 {
+		// r1 should really start at LocalMax but it starts "officially" at KeyMin
+		// which is not addressable.
+		template.Key = keys.LocalMax
+	}
 	template.EndKey = desc.EndKey.AsRawKey()
 
 	return template
@@ -603,34 +599,13 @@ func (r *replicaGCer) send(ctx context.Context, req kvpb.GCRequest) error {
 	n := atomic.AddInt32(&r.count, 1)
 	log.Eventf(ctx, "sending batch %d (%d keys, %d rangekeys)", n, len(req.Keys), len(req.RangeKeys))
 
-	ba := &kvpb.BatchRequest{}
-	// Technically not needed since we're talking directly to the Replica.
-	ba.RangeID = r.repl.Desc().RangeID
-	ba.Timestamp = r.repl.Clock().Now()
-	ba.Add(&req)
-	// Since we are talking directly to the replica, we need to explicitly do
-	// admission control here, as we are bypassing server.Node.
-	var admissionHandle kvadmission.Handle
-	if r.admissionController != nil {
-		ba.AdmissionHeader = gcAdmissionHeader(r.repl.ClusterSettings())
-		ba.Replica.StoreID = r.storeID
-		var err error
-		admissionHandle, err = r.admissionController.AdmitKVWork(ctx, roachpb.SystemTenantID, ba)
-		if err != nil {
-			return err
-		}
-		if mvccGCQueueFullyEnableAC.Get(&r.repl.ClusterSettings().SV) {
-			ctx = admissionHandle.AnnotateCtx(ctx)
-		}
-	}
-	_, writeBytes, pErr := r.repl.SendWithWriteBytes(ctx, ba)
-	defer writeBytes.Release()
-	if r.admissionController != nil {
-		r.admissionController.AdmittedKVWorkDone(admissionHandle, writeBytes)
-	}
-	if pErr != nil {
-		log.VErrEventf(ctx, 2, "%v", pErr.String())
-		return pErr.GoError()
+	var b kv.Batch
+	b.AddRawRequest(&req)
+	b.AdmissionHeader = gcAdmissionHeader(r.repl.ClusterSettings())
+
+	if err := r.repl.store.cfg.DB.Run(ctx, &b); err != nil {
+		log.Infof(ctx, "%s", err)
+		return err
 	}
 	return nil
 }

pkg/kv/kvserver/mvcc_gc_queue_test.go

@@ -1397,6 +1397,9 @@ func TestMVCCGCQueueChunkRequests(t *testing.T) {
 		func(filterArgs kvserverbase.FilterArgs) *kvpb.Error {
 			if _, ok := filterArgs.Req.(*kvpb.GCRequest); ok {
 				atomic.AddInt32(&gcRequests, 1)
+				// Verify that all MVCC GC requests have their admission control header
+				// populated.
+				assert.NotZero(t, filterArgs.AdmissionHdr.CreateTime)
 				return nil
 			}
 			return nil

pkg/kv/kvserver/replica_evaluate.go

@@ -324,13 +324,14 @@ func evaluateBatch(
 		// If a unittest filter was installed, check for an injected error; otherwise, continue.
 		if filter := rec.EvalKnobs().TestingEvalFilter; filter != nil {
 			filterArgs := kvserverbase.FilterArgs{
-				Ctx:     ctx,
-				CmdID:   idKey,
-				Index:   index,
-				Sid:     rec.StoreID(),
-				Req:     args,
-				Version: rec.ClusterSettings().Version.ActiveVersionOrEmpty(ctx).Version,
-				Hdr:     baHeader,
+				Ctx:          ctx,
+				CmdID:        idKey,
+				Index:        index,
+				Sid:          rec.StoreID(),
+				Req:          args,
+				Version:      rec.ClusterSettings().Version.ActiveVersionOrEmpty(ctx).Version,
+				Hdr:          baHeader,
+				AdmissionHdr: ba.AdmissionHeader,
 			}
 			if pErr := filter(filterArgs); pErr != nil {
 				if pErr.GetTxn() == nil {
@@ -356,13 +357,14 @@ func evaluateBatch(
 
 		if filter := rec.EvalKnobs().TestingPostEvalFilter; filter != nil {
 			filterArgs := kvserverbase.FilterArgs{
-				Ctx:   ctx,
-				CmdID: idKey,
-				Index: index,
-				Sid:   rec.StoreID(),
-				Req:   args,
-				Hdr:   baHeader,
-				Err:   err,
+				Ctx:          ctx,
+				CmdID:        idKey,
+				Index:        index,
+				Sid:          rec.StoreID(),
+				Req:          args,
+				Hdr:          baHeader,
+				AdmissionHdr: ba.AdmissionHeader,
+				Err:          err,
 			}
 			if pErr := filter(filterArgs); pErr != nil {
 				if pErr.GetTxn() == nil {

pkg/settings/registry.go

@@ -259,6 +259,7 @@ var retiredSettings = map[InternalKey]struct{}{
 
 	// removed as of 25.2
 	"kv.snapshot_receiver.excise.enabled":                    {},
+	"kv.mvcc_gc.queue_kv_admission_control.enabled":          {},
 	"sql.catalog.experimental_use_session_based_leasing":     {},
 	"bulkio.backup.merge_file_buffer_size":                   {},
 	"changefeed.new_webhook_sink_enabled":                    {},