Refactor/single process multiple services

.golangci.yml

@@ -1,4 +1,7 @@
 version: "2"
+run:
+  build-tags:
+    - endtoendtests
 linters:
   enable:
     - exhaustive

Makefile

@@ -516,6 +516,17 @@ start-postgres: ## Run the PostgreSQL 16 docker container
 	@docker run --rm --name postgres -p 5432:5432 -d -e POSTGRES_PASSWORD=password -e POSTGRES_DB=rollupsdb -v $(CURDIR)/test/postgres/init-test-db.sh:/docker-entrypoint-initdb.d/init-test-db.sh postgres:18-alpine
 	@$(MAKE) migrate
 
+start-awslocalstack: ## Run the AWS LocalStack docker container
+	@echo "Starting AWS localstack"
+	@docker run --rm --name awslocalstack -p 127.0.0.1:4566:4566 -d -e SERVICES=kms localstack/localstack:4.14.0
+	@echo "Add the following variables to run integration test with AWS services:"
+	@echo "  export AWS_ACCESS_KEY_ID=test"
+	@echo "  export AWS_SECRET_ACCESS_KEY=test"
+	@echo "  export AWS_REGION=us-east-1"
+	@echo "  export AWS_ENDPOINT_URL_KMS=http://localhost:4566"
+	@echo "  export LOCALSTACK_KMS_ENDPOINT=http://localhost:4566"
+	@echo "  export LOCALSTACK_KMS_REQUIRED=true"
+
 start: start-postgres start-devnet ## Start the anvil devnet and PostgreSQL 16 docker containers
 
 stop-devnet: ## Stop the anvil devnet docker container
@@ -524,7 +535,10 @@ stop-devnet: ## Stop the anvil devnet docker container
 stop-postgres: ## Stop the PostgreSQL 16 docker container
 	@docker stop postgres || true
 
-stop: stop-devnet stop-postgres ## Stop all running docker containers
+stop-awslocalstack: ## Stop the AWS LocalStack docker container
+	@docker stop awslocalstack || true
+
+stop: stop-devnet stop-postgres ## Stop the anvil devnet and PostgreSQL 16 docker containers
 
 restart-devnet: ## Restart the anvil devnet docker container
 	@$(MAKE) stop-devnet
@@ -571,14 +585,15 @@ check-license: ## Verify license headers on Go source files
 # Discovery (integration-test-shard-check) lists tests with a plain Go
 # toolchain, so the integration package must stay free of the Cartesi CGo
 # dependency for the check to build on the CI setup runner.
-INTEGRATION_SHARDS := basic quorum prt replay restart withdrawal
+INTEGRATION_SHARDS := basic quorum prt replay restart withdrawal awskms
 
 INTEGRATION_SHARD_basic      := ^Test(EchoAuthority|RejectException|MultiApp|EchoAuthorityStaging)$$
 INTEGRATION_SHARD_quorum     := ^Test(EchoQuorum|SameBlockInputs)$$
 INTEGRATION_SHARD_prt        := ^Test(EchoPrt|RejectExceptionPrt|ForeclosePrt)$$
 INTEGRATION_SHARD_replay     := ^Test(Foreclose|ForecloseReplay|DivergentClaim)$$
 INTEGRATION_SHARD_restart    := ^Test(Restart|SnapshotPolicy)$$
 INTEGRATION_SHARD_withdrawal := ^TestWithdrawalLifecycle$$
+INTEGRATION_SHARD_awskms     := ^TestLocalStackAWSIntegration$$
 
 # -----------------------------------------------------------------------------
 # Node topology axis — orthogonal to shards.
@@ -600,7 +615,7 @@ INTEGRATION_TOPOLOGIES := standalone multiprocess
 NODE_TOPOLOGY ?= standalone
 
 INTEGRATION_SHARDS_standalone   := $(INTEGRATION_SHARDS)
-INTEGRATION_SHARDS_multiprocess := $(INTEGRATION_SHARDS)
+INTEGRATION_SHARDS_multiprocess := $(filter-out awskms,$(INTEGRATION_SHARDS))
 
 # The CI matrix is the set of (shard, topology) cells, encoded "shard:topology".
 INTEGRATION_CELLS := $(foreach t,$(INTEGRATION_TOPOLOGIES),$(foreach s,$(INTEGRATION_SHARDS_$(t)),$(s):$(t)))
@@ -624,6 +639,9 @@ TOPOLOGIES_SELECTED = $(if $(filter all,$(NODE_TOPOLOGY)),$(INTEGRATION_TOPOLOGI
 shards_for = $(filter $(if $(strip $(SHARD)),$(SHARD),$(INTEGRATION_SHARDS_$(1))),$(INTEGRATION_SHARDS_$(1)))
 # run_pattern(topology): the selected shards' -run regexes as one alternation.
 run_pattern = $(subst $(space),|,$(strip $(foreach s,$(call shards_for,$(1)),$(INTEGRATION_SHARD_$(s)))))
+# compose_profiles(topology): activate optional infrastructure required by the
+# selected shards for this topology.
+compose_profiles = $(if $(filter awskms,$(call shards_for,$(1))),awskms,)
 # Selected (shard:topology) cells, for PARALLEL fan-out.
 SELECTED_CELLS = $(foreach t,$(TOPOLOGIES_SELECTED),$(foreach s,$(call shards_for,$(t)),$(s):$(t)))
 # Label for project/log names: the SHARD filter joined by '-', or "all".
@@ -671,6 +689,7 @@ _compose-topology-%:
 		COMPOSE_PROJECT='$(if $(filter rollups-node-integration,$(COMPOSE_PROJECT)),rollups-node-integration-$(SUITE_LABEL)-$*,$(COMPOSE_PROJECT))' \
 		INTEGRATION_LOGS='integration-logs-$(SUITE_LABEL)-$*.txt' \
 		TEST_PATTERN="$$pattern" SHARD_NAME='$(SUITE_LABEL)-$*' NODE_TOPOLOGY='$*' \
+		COMPOSE_PROFILES='$(call compose_profiles,$*)' \
 		GOTESTSUM_FORMAT='$(COMPOSE_TOPOLOGY_GOTESTSUM_FORMAT)' \
 		scripts/compose-integration-run.sh
 
@@ -681,6 +700,7 @@ _compose-cell-%:
 		TEST_PATTERN='$(INTEGRATION_SHARD_$(firstword $(subst :, ,$*)))' \
 		SHARD_NAME='$(firstword $(subst :, ,$*))' \
 		NODE_TOPOLOGY='$(lastword $(subst :, ,$*))' \
+		COMPOSE_PROFILES='$(if $(filter awskms,$(firstword $(subst :, ,$*))),awskms,)' \
 		GOTESTSUM_FORMAT='$(GOTESTSUM_FORMAT)' \
 		scripts/compose-integration-run.sh
 
@@ -714,6 +734,14 @@ integration-test-local: build cartesi-rollups-machine-tool echo-dapp reject-loop
 _local-topology-%:
 	@pattern='$(call run_pattern,$*)'; \
 		if [ -z "$$pattern" ]; then echo "skip: no applicable shards for topology '$*' (SHARD filter excludes all)"; exit 0; fi; \
+		if [ -n "$(filter awskms,$(SHARD))" ]; then \
+			if [ -z "$$LOCALSTACK_KMS_ENDPOINT" ]; then \
+				echo "ERROR: LOCALSTACK_KMS_ENDPOINT is required when SHARD includes awskms." >&2; \
+				echo "Run 'make start-awslocalstack' and export the variables it prints." >&2; \
+				exit 1; \
+			fi; \
+			export LOCALSTACK_KMS_REQUIRED=true; \
+		fi; \
 		cartesi-rollups-cli db init; \
 		test_ports="10000 10001 10002 10003 10004 10005 10006 10011 10012"; \
 		busy_pids="$$(for p in $$test_ports; do lsof -tiTCP:$$p -sTCP:LISTEN 2>/dev/null || true; done | sort -u | tr '\n' ' ')"; \

cmd/cartesi-rollups-advancer/root/root.go

@@ -9,6 +9,8 @@ import (
 	"github.com/cartesi/rollups-node/internal/advancer"
 	"github.com/cartesi/rollups-node/internal/cli"
 	"github.com/cartesi/rollups-node/internal/config"
+	"github.com/cartesi/rollups-node/internal/inspect"
+	"github.com/cartesi/rollups-node/internal/manager"
 	"github.com/cartesi/rollups-node/internal/repository/factory"
 	"github.com/cartesi/rollups-node/internal/version"
 	"github.com/cartesi/rollups-node/pkg/service"
@@ -79,29 +81,58 @@ func run(cmd *cobra.Command, args []string) {
 	ctx, cancel := context.WithTimeout(context.Background(), cfg.MaxStartupTime)
 	defer cancel()
 
-	createInfo := advancer.CreateInfo{
-		CreateInfo: service.CreateInfo{
-			Name:                 config.ServiceAdvancer,
-			LogLevel:             config.ResolveServiceLogLevel(config.ServiceAdvancer, cfg.LogLevel),
-			LogColor:             cfg.LogColor,
-			EnableSignalHandling: true,
-			TelemetryCreate:      true,
-			TelemetryAddress:     cfg.AdvancerTelemetryAddress,
-			PollInterval:         cfg.AdvancerPollingInterval,
+	// Create shared components
+
+	name := config.ServiceAdvancer
+	logger := service.NewLogger(name, cfg.LogLevel, cfg.LogColor)
+
+	repo, err := factory.NewRepositoryFromConnectionString(ctx, cfg.DatabaseConnection.Raw())
+	cli.CheckErr(logger, err)
+	defer repo.Close()
+
+	machineManager := manager.NewMachineManager(
+		repo,
+		logger,
+		cfg.FeatureMachineHashCheckEnabled,
+		cfg.AdvancerInputBatchSize,
+	)
+	defer machineManager.Close()
+
+	// Create factories of services
+
+	factories := []service.FactoryFunction{
+		func(ctx context.Context, sup *service.Supervisor) (service.SupervisedService, error) {
+			return advancer.Create(ctx, &advancer.CreateInfo{
+				Config:     *cfg,
+				Repository: repo,
+				Machines:   machineManager,
+				Supervisor: sup,
+				Logger:     sup.Logger,
+			})
 		},
-		Config: *cfg,
 	}
-	logger := service.NewServiceLogger(&createInfo.CreateInfo)
-	createInfo.CreateInfo.Logger = logger
 
-	var err error
-	createInfo.Repository, err = factory.NewRepositoryFromConnectionString(ctx, cfg.DatabaseConnection.Raw())
-	cli.CheckErr(logger, err)
-	defer createInfo.Repository.Close()
+	if cfg.FeatureInspectEnabled {
+		factories = append(factories,
+			func(ctx context.Context, sup *service.Supervisor) (service.SupervisedService, error) {
+				return inspect.Create(ctx, &inspect.CreateInfo{
+					Config:     *cfg,
+					Repository: repo,
+					Machines:   machineManager,
+				})
+			},
+		)
+	}
 
-	advancerService, err := advancer.Create(ctx, &createInfo)
+	supCfg := &service.SupervisorConfigs{
+		BaseConfigs:          service.BaseConfigs{Name: name, Logger: logger},
+		EnableSignalHandling: true,
+		TelemetryCreate:      true,
+		TelemetryAddress:     cfg.AdvancerTelemetryAddress,
+		Factories:            factories,
+	}
+	sup, err := service.NewSupervisor(ctx, supCfg)
 	cli.CheckErr(logger, err)
-	advancerService.LogConfig(createInfo.Config)
-
-	cli.CheckErr(logger, advancerService.Serve())
+	defer sup.Close()
+	cli.CheckErr(logger, sup.Serve())
 }

cmd/cartesi-rollups-claimer/root/root.go

@@ -11,7 +11,6 @@ import (
 	"github.com/cartesi/rollups-node/internal/config"
 	"github.com/cartesi/rollups-node/internal/repository/factory"
 	"github.com/cartesi/rollups-node/internal/version"
-	"github.com/cartesi/rollups-node/pkg/ethutil"
 	"github.com/cartesi/rollups-node/pkg/service"
 
 	"github.com/spf13/cobra"
@@ -80,41 +79,30 @@ func run(cmd *cobra.Command, args []string) {
 	ctx, cancel := context.WithTimeout(context.Background(), cfg.MaxStartupTime)
 	defer cancel()
 
-	createInfo := claimer.CreateInfo{
-		CreateInfo: service.CreateInfo{
-			Name:                 config.ServiceClaimer,
-			LogLevel:             config.ResolveServiceLogLevel(config.ServiceClaimer, cfg.LogLevel),
-			LogColor:             cfg.LogColor,
-			EnableSignalHandling: true,
-			TelemetryCreate:      true,
-			TelemetryAddress:     cfg.ClaimerTelemetryAddress,
-			PollInterval:         cfg.ClaimerPollingInterval,
-		},
-		Config: *cfg,
-	}
-	logger := service.NewServiceLogger(&createInfo.CreateInfo)
-	createInfo.CreateInfo.Logger = logger
-
-	authOpt, err := config.HTTPAuthorizationOption()
-	cli.CheckErr(logger, err)
-	createInfo.EthConn, err = ethutil.NewEthClient(
-		ctx, cfg.BlockchainHttpEndpoint.Raw(), logger,
-		ethutil.RetryConfig{
-			MaxRetries:     cfg.BlockchainHttpMaxRetries,
-			RetryMinWait:   cfg.BlockchainHttpRetryMinWait,
-			RetryMaxWait:   cfg.BlockchainHttpRetryMaxWait,
-			RequestTimeout: cfg.BlockchainHttpRequestTimeout,
-		}, authOpt)
-	cli.CheckErr(logger, err)
+	name := config.ServiceClaimer
+	logger := service.NewLogger(name, cfg.LogLevel, cfg.LogColor)
 
-	createInfo.Repository, err = factory.NewRepositoryFromConnectionString(ctx, cfg.DatabaseConnection.Raw())
+	repo, err := factory.NewRepositoryFromConnectionString(ctx, cfg.DatabaseConnection.Raw())
 	cli.CheckErr(logger, err)
-	defer createInfo.Repository.Close()
-
-	claimerService, err := claimer.Create(ctx, &createInfo)
-	cli.CheckErr(logger, err)
-	claimerService.LogConfig(createInfo.Config)
-
-	err = claimerService.Serve()
+	defer repo.Close()
+
+	supCfg := &service.SupervisorConfigs{
+		BaseConfigs:          service.BaseConfigs{Name: name, Logger: logger},
+		EnableSignalHandling: true,
+		TelemetryCreate:      true,
+		TelemetryAddress:     cfg.ClaimerTelemetryAddress,
+		Factories: []service.FactoryFunction{
+			func(ctx context.Context, sup *service.Supervisor) (service.SupervisedService, error) {
+				return claimer.Create(ctx, &claimer.CreateInfo{
+					Config:     *cfg,
+					Logger:     sup.Logger,
+					Repository: repo,
+				})
+			},
+		},
+	}
+	sup, err := service.NewSupervisor(ctx, supCfg)
 	cli.CheckErr(logger, err)
+	defer sup.Close()
+	cli.CheckErr(logger, sup.Serve())
 }

cmd/cartesi-rollups-cli/root/deposit/deposit.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cartesi/rollups-node/cmd/cartesi-rollups-cli/util"
 	"github.com/cartesi/rollups-node/internal/cli"
 	"github.com/cartesi/rollups-node/internal/config"
+	"github.com/cartesi/rollups-node/internal/config/auth"
 	"github.com/cartesi/rollups-node/pkg/contracts/iapplication"
 	"github.com/cartesi/rollups-node/pkg/contracts/ierc20errors"
 	"github.com/cartesi/rollups-node/pkg/contracts/ierc20metadata"
@@ -113,7 +114,7 @@ func runERC20(cmd *cobra.Command, args []string) {
 	cobra.CheckErr(err)
 	chainID, err := client.ChainID(ctx)
 	cobra.CheckErr(err)
-	txOpts, err := cli.GetTransactOpts(ctx, chainID)
+	txOptsFactory, err := auth.GetTransactOptsFactory(ctx, chainID)
 	cobra.CheckErr(err)
 
 	if !skipConfirmation {
@@ -124,7 +125,7 @@ func runERC20(cmd *cobra.Command, args []string) {
 			"  token:       %s\n"+
 			"  amount:      %s\n"+
 			"  approve:     %t\n",
-			txOpts.From, appAddr, portalAddr, tokenAddr, amount.String(), approveParam)
+			txOptsFactory.From(), appAddr, portalAddr, tokenAddr, amount.String(), approveParam)
 		confirmed, promptErr := cli.ConfirmPrompt("Do you want to continue?")
 		cobra.CheckErr(promptErr)
 		if !confirmed {
@@ -137,7 +138,7 @@ func runERC20(cmd *cobra.Command, args []string) {
 	if approveParam {
 		token, err := ierc20metadata.NewIERC20Metadata(tokenAddr, client)
 		cobra.CheckErr(err)
-		approveOpts, err := cli.GetTransactOpts(ctx, chainID)
+		approveOpts, err := cli.GetTransactOptsFromFactory(ctx, txOptsFactory)
 		cobra.CheckErr(err)
 		tx, err := token.Approve(approveOpts, portalAddr, amount)
 		cobra.CheckErr(cli.DecorateRevert(err,
@@ -153,7 +154,7 @@ func runERC20(cmd *cobra.Command, args []string) {
 
 	portal, err := ierc20portal.NewIERC20Portal(portalAddr, client)
 	cobra.CheckErr(err)
-	depositOpts, err := cli.GetTransactOpts(ctx, chainID)
+	depositOpts, err := cli.GetTransactOptsFromFactory(ctx, txOptsFactory)
 	cobra.CheckErr(err)
 	tx, err := portal.DepositERC20Tokens(depositOpts, tokenAddr, appAddr, amount, execData)
 	// The revert can come from three layers: the portal itself

cmd/cartesi-rollups-evm-reader/root/root.go

@@ -11,7 +11,6 @@ import (
 	"github.com/cartesi/rollups-node/internal/evmreader"
 	"github.com/cartesi/rollups-node/internal/repository/factory"
 	"github.com/cartesi/rollups-node/internal/version"
-	"github.com/cartesi/rollups-node/pkg/ethutil"
 	"github.com/cartesi/rollups-node/pkg/service"
 
 	"github.com/spf13/cobra"
@@ -80,42 +79,30 @@ func run(cmd *cobra.Command, args []string) {
 	ctx, cancel := context.WithTimeout(context.Background(), cfg.MaxStartupTime)
 	defer cancel()
 
-	createInfo := evmreader.CreateInfo{
-		CreateInfo: service.CreateInfo{
-			Name:                 config.ServiceEvmReader,
-			LogLevel:             config.ResolveServiceLogLevel(config.ServiceEvmReader, cfg.LogLevel),
-			LogColor:             cfg.LogColor,
-			EnableSignalHandling: true,
-			TelemetryCreate:      true,
-			TelemetryAddress:     cfg.EvmReaderTelemetryAddress,
-			PollInterval:         cfg.EvmReaderPollingInterval,
-		},
-		Config: *cfg,
-	}
-	logger := service.NewServiceLogger(&createInfo.CreateInfo)
-	createInfo.CreateInfo.Logger = logger
-
-	var err error
-	authOpt, err := config.HTTPAuthorizationOption()
-	cli.CheckErr(logger, err)
-	createInfo.EthClient, err = ethutil.NewEthClient(
-		ctx, cfg.BlockchainHttpEndpoint.Raw(), logger,
-		ethutil.RetryConfig{
-			MaxRetries:     cfg.BlockchainHttpMaxRetries,
-			RetryMinWait:   cfg.BlockchainHttpRetryMinWait,
-			RetryMaxWait:   cfg.BlockchainHttpRetryMaxWait,
-			RequestTimeout: cfg.BlockchainHttpRequestTimeout,
-		}, authOpt)
-	cli.CheckErr(logger, err)
+	name := config.ServiceEvmReader
+	logger := service.NewLogger(name, cfg.LogLevel, cfg.LogColor)
 
 	repo, err := factory.NewRepositoryFromConnectionString(ctx, cfg.DatabaseConnection.Raw())
 	cli.CheckErr(logger, err)
 	defer repo.Close()
-	createInfo.Repository = repo
 
-	readerService, err := evmreader.Create(ctx, &createInfo)
+	supCfg := &service.SupervisorConfigs{
+		BaseConfigs:          service.BaseConfigs{Name: name, Logger: logger},
+		EnableSignalHandling: true,
+		TelemetryCreate:      true,
+		TelemetryAddress:     cfg.EvmReaderTelemetryAddress,
+		Factories: []service.FactoryFunction{
+			func(ctx context.Context, sup *service.Supervisor) (service.SupervisedService, error) {
+				return evmreader.Create(ctx, &evmreader.CreateInfo{
+					Config:     *cfg,
+					Logger:     sup.Logger,
+					Repository: repo,
+				})
+			},
+		},
+	}
+	sup, err := service.NewSupervisor(ctx, supCfg)
 	cli.CheckErr(logger, err)
-	readerService.LogConfig(createInfo.Config)
-
-	cli.CheckErr(logger, readerService.Serve())
+	defer sup.Close()
+	cli.CheckErr(logger, sup.Serve())
 }