fix: respect max_retries and subscription_timeout

- fixed evm-reader `Run`. It was not respecting
  CARTESI_BLOCKCHAIN_SUBSCRIPTION_TIMEOUT and
  CARTESI_BLOCKCHAIN_HTTP_MAX_RETRIES options.
- Propagate `node` Context and Cancel function to services.

Author: mpolitzer

internal/config/generate/Config.toml

@@ -150,13 +150,6 @@ The default block to be used by EVM Reader and Claimer when requesting new block
 One of 'latest', 'pending', 'safe', 'finalized'"""
 used-by = ["evmreader", "claimer", "node", "prt"]
 
-[blockchain.CARTESI_BLOCKCHAIN_SUBSCRIPTION_TIMEOUT]
-default = "60"
-go-type = "Duration"
-description = """
-Block subscription timeout in seconds."""
-used-by = ["evmreader", "node"]
-
 [rollups.CARTESI_BLOCKCHAIN_HTTP_MAX_RETRIES]
 default = "4"
 go-type = "uint64"

internal/config/generated.go

@@ -72,15 +72,28 @@ type appContracts struct {
 }
 
 func (r *Service) Run(ctx context.Context, ready chan struct{}) error {
-	for {
+	for attempt := uint64(1); ; attempt++ {
 		err := r.watchForNewBlocks(ctx, ready)
-		// If the error is a SubscriptionError, re run watchForNewBlocks
-		// that it will restart the websocket subscription
-		if _, ok := err.(*SubscriptionError); !ok {
+		r.Logger.Error(err.Error())
+
+		if attempt > r.blockchainMaxRetries {
+			r.Logger.Error("Max attempts reached for subscription restart. Exititng",
+				"max_retries", r.blockchainMaxRetries,
+			)
 			return err
 		}
-		r.Logger.Error(err.Error())
-		r.Logger.Info("Restarting subscription")
+
+		r.Logger.Info("Restarting subscription",
+			"attempt", attempt,
+			"remaining", r.blockchainMaxRetries - attempt,
+			"time_between_attempts", r.blockchainSubscriptionRetryInterval,
+		)
+
+		// sleep or cancel
+		select {
+		case <-ctx.Done():
+		case <-time.After(r.blockchainSubscriptionRetryInterval):
+		}
 	}
 }
 

internal/evmreader/evmreader.go

@@ -145,13 +145,15 @@ func (s *EvmReaderSuite) SetupTest() {
 	s.contractFactory = newMockAdapterFactory().SetupDefaultBehavior(s.applicationContract1, s.applicationContract2, s.inputBox)
 
 	s.evmReader = &Service{
-		client:             s.client,
-		wsClient:           s.wsClient,
-		repository:         s.repository,
-		defaultBlock:       DefaultBlock_Latest,
-		adapterFactory:     s.contractFactory,
-		hasEnabledApps:     true,
-		inputReaderEnabled: true,
+		client:                              s.client,
+		wsClient:                            s.wsClient,
+		repository:                          s.repository,
+		defaultBlock:                        DefaultBlock_Latest,
+		adapterFactory:                      s.contractFactory,
+		hasEnabledApps:                      true,
+		inputReaderEnabled:                  true,
+		blockchainMaxRetries:                0,
+		blockchainSubscriptionRetryInterval: time.Second,
 	}
 
 	logLevel, err := config.GetLogLevel()

internal/evmreader/evmreader_test.go

@@ -527,13 +527,15 @@ func (s *EvmReaderSuite) setupOutputMismatchTest() {
 	s.contractFactory = newMockAdapterFactory()
 
 	s.evmReader = &Service{
-		client:             s.client,
-		wsClient:           s.wsClient,
-		repository:         s.repository,
-		defaultBlock:       DefaultBlock_Latest,
-		adapterFactory:     s.contractFactory,
-		hasEnabledApps:     true,
-		inputReaderEnabled: true,
+		client:                              s.client,
+		wsClient:                            s.wsClient,
+		repository:                          s.repository,
+		defaultBlock:                        DefaultBlock_Latest,
+		adapterFactory:                      s.contractFactory,
+		hasEnabledApps:                      true,
+		inputReaderEnabled:                  true,
+		blockchainMaxRetries:                0,
+		blockchainSubscriptionRetryInterval: time.Second,
 	}
 
 	logLevel, err := config.GetLogLevel()

internal/evmreader/output_test.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"time"
 
 	"github.com/cartesi/rollups-node/internal/config"
 	. "github.com/cartesi/rollups-node/internal/model"
@@ -30,14 +31,16 @@ type CreateInfo struct {
 type Service struct {
 	service.Service
 
-	client             EthClientInterface
-	wsClient           EthClientInterface
-	adapterFactory     AdapterFactory
-	repository         EvmReaderRepository
-	chainId            uint64
-	defaultBlock       DefaultBlock
-	hasEnabledApps     bool
-	inputReaderEnabled bool
+	client                              EthClientInterface
+	wsClient                            EthClientInterface
+	adapterFactory                      AdapterFactory
+	repository                          EvmReaderRepository
+	chainId                             uint64
+	defaultBlock                        DefaultBlock
+	hasEnabledApps                      bool
+	inputReaderEnabled                  bool
+	blockchainMaxRetries                uint64
+	blockchainSubscriptionRetryInterval time.Duration
 }
 
 const EvmReaderConfigKey = "evm-reader"
@@ -99,6 +102,8 @@ func Create(ctx context.Context, c *CreateInfo) (*Service, error) {
 		return nil, fmt.Errorf("NodeConfig chainId mismatch: network %d != config %d",
 			chainId.Uint64(), nodeConfig.ChainID)
 	}
+	s.blockchainMaxRetries = c.Config.BlockchainHttpMaxRetries
+	s.blockchainSubscriptionRetryInterval = c.Config.BlockchainHttpRetryMinWait
 
 	s.client = c.EthClient
 	s.wsClient = c.EthWsClient
@@ -140,7 +145,10 @@ func (s *Service) Tick() []error {
 
 func (s *Service) Serve() error {
 	ready := make(chan struct{}, 1)
-	go s.Run(s.Context, ready)
+	go func() {
+		s.Run(s.Context, ready)
+		s.Service.Stop(false)
+	}()
 	return s.Service.Serve()
 }
 

internal/evmreader/service.go

@@ -155,6 +155,8 @@ func newEVMReader(ctx context.Context, c *CreateInfo, s *Service) service.IServi
 	readerArgs := evmreader.CreateInfo{
 		CreateInfo: service.CreateInfo{
 			Name:                 "evm-reader",
+			Context:              s.Context,
+			Cancel:               s.Cancel,
 			LogLevel:             c.Config.LogLevel,
 			LogColor:             c.Config.LogColor,
 			EnableSignalHandling: false,
@@ -179,6 +181,8 @@ func newAdvancer(ctx context.Context, c *CreateInfo, s *Service) service.IServic
 	advancerArgs := advancer.CreateInfo{
 		CreateInfo: service.CreateInfo{
 			Name:                 "advancer",
+			Context:              s.Context,
+			Cancel:               s.Cancel,
 			LogLevel:             c.Config.LogLevel,
 			LogColor:             c.Config.LogColor,
 			EnableSignalHandling: false,
@@ -202,6 +206,8 @@ func newValidator(ctx context.Context, c *CreateInfo, s *Service) service.IServi
 	validatorArgs := validator.CreateInfo{
 		CreateInfo: service.CreateInfo{
 			Name:                 "validator",
+			Context:              s.Context,
+			Cancel:               s.Cancel,
 			LogLevel:             c.Config.LogLevel,
 			LogColor:             c.Config.LogColor,
 			EnableSignalHandling: false,
@@ -225,6 +231,8 @@ func newClaimer(ctx context.Context, c *CreateInfo, s *Service) service.IService
 	claimerArgs := claimer.CreateInfo{
 		CreateInfo: service.CreateInfo{
 			Name:                 "claimer",
+			Context:              s.Context,
+			Cancel:               s.Cancel,
 			LogLevel:             c.Config.LogLevel,
 			LogColor:             c.Config.LogColor,
 			EnableSignalHandling: false,
@@ -249,6 +257,8 @@ func newJsonrpc(ctx context.Context, c *CreateInfo, s *Service) service.IService
 	jsonrpcArgs := jsonrpc.CreateInfo{
 		CreateInfo: service.CreateInfo{
 			Name:                 "jsonrpc",
+			Context:              s.Context,
+			Cancel:               s.Cancel,
 			LogLevel:             c.Config.LogLevel,
 			LogColor:             c.Config.LogColor,
 			EnableSignalHandling: false,
@@ -271,6 +281,8 @@ func newPrt(ctx context.Context, c *CreateInfo, s *Service) service.IService {
 	prtArgs := prt.CreateInfo{
 		CreateInfo: service.CreateInfo{
 			Name:                 "prt",
+			Context:              s.Context,
+			Cancel:               s.Cancel,
 			LogLevel:             c.Config.LogLevel,
 			LogColor:             c.Config.LogColor,
 			EnableSignalHandling: false,

internal/node/node.go

@@ -103,6 +103,7 @@ type CreateInfo struct {
 	Impl                 ServiceImpl
 	ServeMux             *http.ServeMux
 	Context              context.Context
+	Cancel               context.CancelFunc
 }
 
 // Service stores runtime information.
@@ -151,7 +152,10 @@ func Create(ctx context.Context, c *CreateInfo, s *Service) error {
 		s.Context = c.Context
 	}
 	if s.Cancel == nil {
-		s.Context, s.Cancel = context.WithCancel(c.Context)
+		if c.Cancel == nil {
+			s.Context, c.Cancel = context.WithCancel(c.Context)
+		}
+		s.Cancel = c.Cancel
 	}
 
 	// ticker
@@ -246,6 +250,7 @@ func (s *Service) Stop(force bool) []error {
 	elapsed := time.Since(start)
 
 	s.Running.Store(false)
+	s.Cancel()
 	if len(errs) > 0 {
 		s.Logger.Error("Stop",
 			"force", force,