feat: add remaining Rust services to Go binary

build/docker-compose.yml

@@ -11,6 +11,7 @@ services:
     entrypoint: ["cartesi-rollups-node", "validator"]
     ports:
       - "4000:4000"
+      - "5005:5005"
     depends_on:
       hardhat:
         condition: service_healthy
@@ -29,6 +30,26 @@ services:
       CARTESI_LOG_LEVEL: info
       POSTGRES_ENDPOINT: postgres://postgres:password@database:5432/postgres
 
+      #Advance Runner
+      SERVER_MANAGER_ENDPOINT: http://server_manager:5001
+      PROVIDER_HTTP_ENDPOINT: http://hardhat:8545
+      SESSION_ID: default_rollups_id
+      SNAPSHOT_DIR: /var/opt/cartesi/machine-snapshots
+      SNAPSHOT_LATEST: /var/opt/cartesi/machine-snapshots/latest
+
+      #Dispatcher
+      DISPATCHER_HTTP_SERVER_PORT: 8081
+      RD_DAPP_DEPLOYMENT_FILE: /deployments/localhost/dapp.json
+      RD_ROLLUPS_DEPLOYMENT_FILE: /opt/cartesi/share/deployments/localhost.json
+      RD_EPOCH_DURATION: 86400
+      SC_GRPC_ENDPOINT: http://0.0.0.0:50051
+      SC_DEFAULT_CONFIRMATIONS: 1
+      TX_PROVIDER_HTTP_ENDPOINT: http://hardhat:8545
+      AUTH_MNEMONIC: "test test test test test test test test test test test junk"
+      TX_CHAIN_ID: 31337
+      TX_CHAIN_IS_LEGACY: ${TX_LEGACY:-false}
+      TX_DEFAULT_CONFIRMATIONS: 2
+
       #GraphQL Server
       GRAPHQL_HEALTHCHECK_PORT: 8082
       GRAPHQL_HOST: "0.0.0.0"
@@ -40,6 +61,19 @@ services:
       REDIS_ENDPOINT: redis://redis:6379
       CHAIN_ID: 31337
 
+      #Inspect Server
+      INSPECT_SERVER_HEALTHCHECK_PORT: 8084
+      INSPECT_SERVER_ADDRESS: 0.0.0.0:5005
+      SERVER_MANAGER_ADDRESS: server_manager:5001
+
+      #State Server
+      SS_SERVER_ADDRESS: 0.0.0.0:50051
+      SF_GENESIS_BLOCK: 0x1
+      SF_SAFETY_MARGIN: 1
+      BH_HTTP_ENDPOINT: http://hardhat:8545
+      BH_WS_ENDPOINT: ws://hardhat:8545
+      BH_BLOCK_TIMEOUT: 8
+
     volumes:
       - machine:/var/opt/cartesi/machine-snapshots
       - blockchain-data:/opt/cartesi/share/deployments:ro

cmd/cartesi-rollups-node/main.go

@@ -4,6 +4,7 @@
 package main
 
 import (
+	"context"
 	"os"
 
 	"github.com/cartesi/rollups-node/internal/logger"
@@ -14,7 +15,8 @@ func main() {
 	_, enableTimestamp := os.LookupEnv("CARTESI_LOG_ENABLE_TIMESTAMP")
 	logger.Init(logLevel, enableTimestamp)
 
-	if err := rootCmd.Execute(); err != nil {
+	ctx := context.Background()
+	if err := rootCmd.ExecuteContext(ctx); err != nil {
 		logger.Error.Panic(err)
 	}
 }

cmd/cartesi-rollups-node/validator.go

@@ -17,9 +17,13 @@ var validator = &cobra.Command{
 
 func runValidatorNode(cmd *cobra.Command, args []string) {
 	validatorServices := []services.Service{
+		services.StateServer, // must be initialized before Dispatcher
+		services.AdvanceRunner,
 		services.GraphQLServer,
 		services.Indexer,
+		services.InspectServer,
+		services.Dispatcher,
 	}
 
-	services.Run(validatorServices)
+	services.Run(cmd.Context(), validatorServices)
 }

internal/services/service.go

@@ -7,8 +7,10 @@ package services
 import (
 	"context"
 	"fmt"
+	"net"
 	"os"
 	"os/exec"
+	"strings"
 	"syscall"
 	"time"
 
@@ -22,14 +24,23 @@ type Service interface {
 	// Start a service that will run until completion or until the context is
 	// canceled
 	Start(ctx context.Context) error
+
+	// Block until the service is ready or context is canceled
+	Ready(ctx context.Context) error
 }
 
-const DefaultServiceTimeout = 15 * time.Second
+const (
+	DefaultServiceTimeout  = 15 * time.Second
+	DefaultDialInterval    = 100 * time.Millisecond
+	DefaultStateServerPort = "50051"
+	DefaultHealthcheckPort = "8080"
+)
 
 // simpleService implements the context cancelation logic of the Service interface
 type simpleService struct {
-	serviceName string
-	binaryName  string
+	serviceName     string
+	binaryName      string
+	healthcheckPort string
 }
 
 func (s simpleService) Start(ctx context.Context) error {
@@ -57,21 +68,43 @@ func (s simpleService) Start(ctx context.Context) error {
 	return nil
 }
 
+// Ready blocks until the service is ready or the context is canceled.
+//
+// A service is considered ready when it is possible to establish a connection
+// to its healthcheck endpoint.
+func (s simpleService) Ready(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			conn, err := net.Dial("tcp", fmt.Sprintf("0.0.0.0:%s", s.healthcheckPort))
+			if err == nil {
+				logger.Debug.Printf("%s is ready\n", s.name)
+				conn.Close()
+				return nil
+			}
+			time.Sleep(DefaultDialInterval)
+		}
+	}
+}
+
 func (s simpleService) String() string {
 	return s.serviceName
 }
 
 // The Run function serves as a very simple supervisor: it will start all the
 // services provided to it and will run until the first of them finishes. Next
 // it will try to stop the remaining services or timeout if they take too long
-func Run(services []Service) {
+func Run(ctx context.Context, services []Service) {
 	if len(services) == 0 {
 		logger.Error.Panic("there are no services to run")
 	}
 
 	// start services
-	ctx, cancel := context.WithCancel(context.Background())
-	exit := make(chan struct{})
+	startedServices := 0
+	ctx, cancel := context.WithCancel(ctx)
+	exit := make(chan struct{}, len(services))
 	for _, service := range services {
 		service := service
 		go func() {
@@ -84,6 +117,17 @@ func Run(services []Service) {
 			}
 			exit <- struct{}{}
 		}()
+
+		// wait for service to be ready or stop all services if it times out
+		readyCtx, readyCancel := context.WithTimeout(ctx, DefaultServiceTimeout)
+		defer readyCancel()
+		if err := service.Ready(readyCtx); err != nil {
+			msg := "main: service '%v' failed to be ready with error: %v. Exiting\n"
+			logger.Error.Printf(msg, service.name, err)
+			exit <- struct{}{}
+			break
+		}
+		startedServicesCount++
 	}
 
 	// wait for first service to exit
@@ -94,7 +138,7 @@ func Run(services []Service) {
 	wait := make(chan struct{})
 	go func() {
 		cancel()
-		for i := 0; i < len(services)-1; i++ {
+		for i := 0; i < startedServices; i++ {
 			<-exit
 		}
 		wait <- struct{}{}
@@ -104,17 +148,70 @@ func Run(services []Service) {
 	case <-wait:
 		logger.Info.Println("main: all services were shutdown")
 	case <-time.After(DefaultServiceTimeout):
-		logger.Warning.Println("main: exited after timeout")
+		logger.Warning.Println("main: exited after a timeout")
 	}
 }
 
+func healthcheckPort(serviceName string) string {
+	if serviceName == "state-server" {
+		if address, ok := os.LookupEnv("SS_SERVER_ADDRESS"); ok {
+			split := strings.Split(address, ":")
+			if len(split) > 1 {
+				return split[1]
+			}
+		}
+		return DefaultStateServerPort
+	}
+
+	env := healthcheckEnv(serviceName)
+	if port, ok := os.LookupEnv(env); ok {
+		if serviceName == "state-server" {
+			split := strings.Split(port, ":")
+			return split[1]
+		}
+		return port
+	} else {
+		return DefaultHealthcheckPort
+	}
+}
+
+func healthcheckEnv(serviceName string) string {
+	if serviceName == "dispatcher" {
+		return "DISPATCHER_HTTP_SERVER_PORT"
+	}
+	normalizedName := strings.Replace(serviceName, "-", "_", -1)
+	return fmt.Sprintf("%s_HEALTHCHECK_PORT", strings.ToUpper(normalizedName))
+}
+
 var (
+	AdvanceRunner = simpleService{
+		serviceName:     "advance-runner",
+		binaryName:      "cartesi-rollups-advance-runner",
+		healthcheckPort: healthcheckPort("advance-runner"),
+	}
+	Dispatcher = simpleService{
+		serviceName:     "dispatcher",
+		binaryName:      "cartesi-rollups-dispatcher",
+		healthcheckPort: healthcheckPort("dispatcher"),
+	}
 	GraphQLServer Service = simpleService{
-		serviceName: "graphql-server",
-		binaryName:  "cartesi-rollups-graphql-server",
+		serviceName:     "graphql-server",
+		binaryName:      "cartesi-rollups-graphql-server",
+		healthcheckPort: healthcheckPort("graphql-server"),
 	}
 	Indexer Service = simpleService{
-		serviceName: "indexer",
-		binaryName:  "cartesi-rollups-indexer",
+		serviceName:     "indexer",
+		binaryName:      "cartesi-rollups-indexer",
+		healthcheckPort: healthcheckPort("indexer"),
+	}
+	InspectServer = simpleService{
+		serviceName:     "inspect-server",
+		binaryName:      "cartesi-rollups-inspect-server",
+		healthcheckPort: healthcheckPort("inspect-server"),
+	}
+	StateServer = simpleService{
+		serviceName:     "state-server",
+		binaryName:      "cartesi-rollups-state-server",
+		healthcheckPort: healthcheckPort("state-server"),
 	}
 )