Merge #148352

148352: kvserver: register `SideTransport` service with DRPC server r=cthumuluru-crdb a=shubhamdhama

Enable the `SideTransport` service on the DRPC server in addition to gRPC. 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

Co-authored-by: Shubham Dhama <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/closedts/sidetransport/receiver.go

@@ -76,8 +76,23 @@ func NewReceiver(
 	return r
 }
 
+func (s *Receiver) AsDRPCServer() ctpb.DRPCSideTransportServer {
+	return (*drpcReceiver)(s)
+}
+
+type drpcReceiver Receiver
+
+// PushUpdates is the streaming RPC handler.
+func (s *drpcReceiver) PushUpdates(stream ctpb.DRPCSideTransport_PushUpdatesStream) error {
+	return (*Receiver)(s).pushUpdates(stream)
+}
+
 // PushUpdates is the streaming RPC handler.
 func (s *Receiver) PushUpdates(stream ctpb.SideTransport_PushUpdatesServer) error {
+	return s.pushUpdates(stream)
+}
+
+func (s *Receiver) pushUpdates(stream ctpb.RPCSideTransport_PushUpdatesStream) error {
 	// Create a steam to service this connection. The stream will call back into
 	// the Receiver through onFirstMsg to register itself once it finds out the
 	// sender's node id.
@@ -288,7 +303,7 @@ func (r *incomingStream) Run(
 	ctx context.Context,
 	stopper *stop.Stopper,
 	// The gRPC stream with incoming messages.
-	stream ctpb.SideTransport_PushUpdatesServer,
+	stream ctpb.RPCSideTransport_PushUpdatesStream,
 ) error {
 	// We have to do the stream processing on a separate goroutine because Recv()
 	// is blocking, with no way to interrupt it other than returning from the RPC

pkg/server/server.go

@@ -995,6 +995,9 @@ func NewServer(cfg Config, stopper *stop.Stopper) (serverctl.ServerStartupInterf
 		return nil, err
 	}
 	ctpb.RegisterSideTransportServer(grpcServer.Server, ctReceiver)
+	if err := ctpb.DRPCRegisterSideTransport(drpcServer, ctReceiver.AsDRPCServer()); err != nil {
+		return nil, err
+	}
 
 	// Create blob service for inter-node file sharing.
 	blobService, err := blobs.NewBlobService(cfg.ExternalIODir)