[Merged by Bors] - Make grpc servers' start methods be synchronous

api/grpcserver/grpc.go

@@ -4,6 +4,7 @@
 	"net"
 	"time"
 
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/reflection"
@@ -21,6 +22,7 @@
 	Listener   string
 	logger     log.Logger
 	GrpcServer *grpc.Server
+	grp        errgroup.Group
 }
 
 // New creates and returns a new Server with port and interface.
@@ -34,7 +36,7 @@
 }
 
 // Start starts the server.
-func (s *Server) Start() <-chan struct{} {
+func (s *Server) Start() error {
 	s.logger.With().Info("starting grpc server",
 		log.String("address", s.Listener),
 		log.Array("services", log.ArrayMarshalerFunc(func(encoder log.ArrayEncoder) error {
@@ -44,35 +46,27 @@
 			return nil
 		})),
 	)
-
-	started := make(chan struct{})
-	go s.startInternal(started)
-
-	return started
-}
-
-// Blocking, should be called in a goroutine.
-func (s *Server) startInternal(started chan<- struct{}) {
 	lis, err := net.Listen("tcp", s.Listener)
 	if err != nil {
 		s.logger.Error("error listening: %v", err)
-		return
+		return err
 	}
 	reflection.Register(s.GrpcServer)
-	s.logger.Info("starting new grpc server on %s", s.Listener)
-	close(started)
-	if err := s.GrpcServer.Serve(lis); err != nil {
-		s.logger.Error("error stopping grpc server: %v", err)
-	}
+	s.grp.Go(func() error {
+		if err := s.GrpcServer.Serve(lis); err != nil {
+			s.logger.Error("error stopping grpc server: %v", err)
+			return err
+		}
+		return nil
+	})
+	return nil
 }
 
 // Close stops the server.
 func (s *Server) Close() error {
 	s.logger.Info("stopping the grpc server")
 	s.GrpcServer.Stop()
-
-	// We don't return any errors but we want to conform to io.Closer so return a nil error
-	return nil
+	return s.grp.Wait()
 }
 
 // ServerOptions are shared by all grpc servers.

api/grpcserver/grpcserver_test.go

@@ -491,18 +491,11 @@ func launchServer(tb testing.TB, cfg Config, services ...ServiceAPI) func() {
 	}
 
 	// start gRPC and json servers
-	grpcStarted := grpcService.Start()
-	jsonStarted := jsonService.StartService(context.Background(), services...)
-
-	timer := time.NewTimer(3 * time.Second)
-	defer timer.Stop()
-
-	// wait for server to be ready (critical on CI)
-	for _, ch := range []<-chan struct{}{grpcStarted, jsonStarted} {
-		select {
-		case <-ch:
-		case <-timer.C:
-		}
+	err := grpcService.Start()
+	require.NoError(tb, err)
+	if len(services) > 0 {
+		err = jsonService.StartService(context.Background(), services...)
+		require.NoError(tb, err)
 	}
 
 	return func() {

api/grpcserver/http_server.go

@@ -4,11 +4,12 @@
 	"context"
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
-	"sync"
 
 	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	pb "github.com/spacemeshos/api/release/go/spacemesh/v1"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/spacemeshos/go-spacemesh/log"
 )
@@ -18,9 +19,9 @@
 type JSONHTTPServer struct {
 	logger log.Logger
 
-	mu       sync.RWMutex
 	listener string
 	server   *http.Server
+	grp      errgroup.Group
 }
 
 // NewJSONHTTPServer creates a new json http server.
@@ -34,41 +35,32 @@
 // Shutdown stops the server.
 func (s *JSONHTTPServer) Shutdown(ctx context.Context) error {
 	s.logger.Debug("stopping json-http service...")
-	server := s.getServer()
-	if server != nil {
-		err := server.Shutdown(ctx)
+	if s.server != nil {
+		err := s.server.Shutdown(ctx)
 		if errors.Is(err, http.ErrServerClosed) {
 			return nil
 		}
 		if err != nil {
 			return fmt.Errorf("shutdown: %w", err)
 		}
 	}
-
-	return nil
+	err := s.grp.Wait()
+	if errors.Is(err, http.ErrServerClosed) {
+		return nil
+	}
+	return err
 }
 
 // StartService starts the json api server and listens for status (started, stopped).
 func (s *JSONHTTPServer) StartService(
 	ctx context.Context,
 	services ...ServiceAPI,
-) <-chan struct{} {
-	started := make(chan struct{})
-
-	// This will block, so run it in a goroutine
-	go s.startInternal(
-		ctx,
-		started,
-		services...)
-
-	return started
-}
-
-func (s *JSONHTTPServer) startInternal(
-	ctx context.Context,
-	started chan<- struct{},
-	services ...ServiceAPI,
-) {
+) error {
+	// At least one service must be enabled
+	if len(services) == 0 {
+		s.logger.Error("not starting grpc gateway service; at least one service must be enabled")
+		return errors.New("no services provided")
+	}
 	ctx, cancel := context.WithCancel(ctx)
 
 	// This will close all downstream connections when the server closes
@@ -96,38 +88,27 @@
 		}
 		if err != nil {
 			s.logger.Error("registering %T with grpc gateway failed with %v", svc, err)
+			return err
 		}
 		serviceCount++
 	}
 
-	close(started)
+	s.logger.With().Info("starting grpc gateway server", log.String("address", s.listener))
 
-	// At least one service must be enabled
-	if serviceCount == 0 {
-		s.logger.Error("not starting grpc gateway service; at least one service must be enabled")
-		return
+	lis, err := net.Listen("tcp", s.listener)
+	if err != nil {
+		s.logger.Error("error listening: %v", err)
+		return err
 	}
-
-	s.logger.With().Info("starting grpc gateway server", log.String("address", s.listener))
-	s.setServer(&http.Server{
-		Addr:    s.listener,
+	s.server = &http.Server{
 		Handler: mux,
+	}
+	s.grp.Go(func() error {
+		if err := s.server.Serve(lis); err != nil {
+			s.logger.Error("error from grpc http server: %v", err)
+			return err
+		}
+		return nil
 	})
-
-	// This will block
-	s.logger.Error("error from grpc http listener: %v", s.getServer().ListenAndServe())
-}
-
-func (s *JSONHTTPServer) getServer() *http.Server {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
-	return s.server
-}
-
-func (s *JSONHTTPServer) setServer(server *http.Server) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	s.server = server
+	return nil
 }

cmd/bootstrapper/generator_test.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -139,23 +140,18 @@ func launchServer(tb testing.TB, cdb *datastore.CachedDB) func() {
 
 	pb.RegisterMeshServiceServer(grpcService.GrpcServer, s)
 	// start gRPC and json servers
-	grpcStarted := grpcService.Start()
-	jsonStarted := jsonService.StartService(context.Background(), s)
-
-	timer := time.NewTimer(3 * time.Second)
-	defer timer.Stop()
-
-	// wait for server to be ready (critical on CI)
-	for _, ch := range []<-chan struct{}{grpcStarted, jsonStarted} {
-		select {
-		case <-ch:
-		case <-timer.C:
-		}
-	}
+	err := grpcService.Start()
+	require.NoError(tb, err)
+	err = jsonService.StartService(context.Background(), s)
+	require.NoError(tb, err)
 
 	return func() {
-		require.NoError(tb, jsonService.Shutdown(context.Background()))
-		_ = grpcService.Close()
+		err := jsonService.Shutdown(context.Background())
+		if !errors.Is(err, http.ErrServerClosed) {
+			require.NoError(tb, err)
+		}
+		err = grpcService.Close()
+		require.NoError(tb, err)
 	}
 }
 

node/bad_peer_test.go

@@ -36,16 +36,22 @@ func TestPeerDisconnectForMessageResultValidationReject(t *testing.T) {
 	conf1.DataDirParent = t.TempDir()
 	conf1.FileLock = filepath.Join(conf1.DataDirParent, "LOCK")
 	conf1.P2P.Listen = "/ip4/127.0.0.1/tcp/0"
+	// We setup the api to listen on an OS assigned port, which avoids the second instance getting stuck when
+	conf1.API.PublicListener = "0.0.0.0:0"
+	conf1.API.PrivateListener = "0.0.0.0:0"
 	app1, err := NewApp(&conf1)
 	require.NoError(t, err)
 	conf2 := config.DefaultTestConfig()
+
 	// We need to copy the genesis config to ensure that both nodes share the
 	// same gnenesis ID, otherwise they will not be able to connect to each
 	// other.
 	*conf2.Genesis = *conf1.Genesis
 	conf2.DataDirParent = t.TempDir()
 	conf2.FileLock = filepath.Join(conf2.DataDirParent, "LOCK")
 	conf2.P2P.Listen = "/ip4/127.0.0.1/tcp/0"
+	conf2.API.PublicListener = "0.0.0.0:0"
+	conf2.API.PrivateListener = "0.0.0.0:0"
 	app2, err := NewApp(&conf2)
 	require.NoError(t, err)
 
@@ -54,13 +60,13 @@ func TestPeerDisconnectForMessageResultValidationReject(t *testing.T) {
 		app1.Cleanup(ctx)
 		app2.Cleanup(ctx)
 	})
-	g := errgroup.Group{}
+	g, grpContext := errgroup.WithContext(ctx)
 	g.Go(func() error {
-		return app1.Start(ctx)
+		return app1.Start(grpContext)
 	})
 	<-app1.Started()
 	g.Go(func() error {
-		return app2.Start(ctx)
+		return app2.Start(grpContext)
 	})
 	<-app2.Started()
 

node/node.go

@@ -1109,10 +1109,14 @@
 		app.jsonAPIService.StartService(ctx, public...)
 	}
 	if app.grpcPublicService != nil {
-		app.grpcPublicService.Start()
+		if err := app.grpcPublicService.Start(); err != nil {
+			return err
+		}
 	}
 	if app.grpcPrivateService != nil {
-		app.grpcPrivateService.Start()
+		if err := app.grpcPrivateService.Start(); err != nil {
+			return err
+		}
 	}
 	return nil
 }