kv,server: extract RangeFeed and register with DRPC server

This is a follow-up to #145195, where we extracted KVBatch from the
Internal service. Here, we do the same with RangeFeed. The TL;DR is
that smaller services are easier to maintain and can be more smoothly
migrated to DRPC.

We also enable this service on the DRPC server. This is controlled by
`rpc.experimental_drpc.enabled` (off by default).

This change is part of a series and is similar to: #146926

Note: This only registers the service; the client is not updated to use the
DRPC client, so this service will not have any functional effect.

Epic: CRDB-48925
Release note: None

Author: cthumuluru-crdb

pkg/kv/kvpb/api.proto

@@ -3883,3 +3883,8 @@ message ScanStats {
 // UsedFollowerRead indicates whether at least some reads were served by the
 // follower replicas.
 message UsedFollowerRead {}
+
+// RangeFeed service implemented by nodes for KV API requests.
+service RangeFeed {
+  rpc MuxRangeFeed       (stream RangeFeedRequest)   returns (stream MuxRangeFeedEvent)       {}
+}

pkg/server/node.go

@@ -1918,6 +1918,16 @@ func (n *kvBatchServer) BatchStream(stream kvpb.DRPCKVBatch_BatchStreamStream) e
 	return (*Node)(n).batchStreamImpl(stream)
 }
 
+type kvRangeFeedServer Node
+
+func (n *Node) AsDRPCRangeFeedServer() kvpb.DRPCRangeFeedServer {
+	return (*kvRangeFeedServer)(n)
+}
+
+func (n *kvRangeFeedServer) MuxRangeFeed(stream kvpb.DRPCRangeFeed_MuxRangeFeedStream) error {
+	return (*Node)(n).muxRangeFeed(stream)
+}
+
 // spanForRequest is the retval of setupSpanForIncomingRPC. It groups together a
 // few variables needed when finishing an RPC's span.
 //
@@ -2092,7 +2102,7 @@ func (n *Node) RangeLookup(
 // MuxRangeFeedServer (default grpc.Stream) is not safe for concurrent calls to
 // Send.
 type lockedMuxStream struct {
-	wrapped kvpb.Internal_MuxRangeFeedServer
+	wrapped kvpb.RPCInternal_MuxRangeFeedStream
 	sendMu  syncutil.Mutex
 	metrics *rangefeed.LockedMuxStreamMetrics
 }
@@ -2188,6 +2198,11 @@ func (n *Node) defaultRangefeedConsumerID() int64 {
 
 // MuxRangeFeed implements the roachpb.InternalServer interface.
 func (n *Node) MuxRangeFeed(muxStream kvpb.Internal_MuxRangeFeedServer) error {
+	return n.muxRangeFeed(muxStream)
+}
+
+// MuxRangeFeed implements the roachpb.InternalServer interface.
+func (n *Node) muxRangeFeed(muxStream kvpb.RPCInternal_MuxRangeFeedStream) error {
 	lockedMuxStream := &lockedMuxStream{
 		wrapped: muxStream,
 		metrics: n.metrics.LockedMuxStreamMetrics,

pkg/server/server.go

@@ -986,6 +986,9 @@ func NewServer(cfg Config, stopper *stop.Stopper) (serverctl.ServerStartupInterf
 	if err := kvpb.DRPCRegisterKVBatch(drpcServer, node.AsDRPCKVBatchServer()); err != nil {
 		return nil, err
 	}
+	if err := kvpb.DRPCRegisterRangeFeed(drpcServer, node.AsDRPCRangeFeedServer()); err != nil {
+		return nil, err
+	}
 	if err := kvpb.DRPCRegisterTenantService(drpcServer, node.AsDRPCTenantServiceServer()); err != nil {
 		return nil, err
 	}