fix(evmreader): EVM Reader is ready only when polling sucessfully

Author: renatomaia

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

@@ -125,7 +125,7 @@ func run(cmd *cobra.Command, args []string) {
 	}
 
 	supCfg := &service.SupervisorConfigs{
-		BaseConfigs: service.BaseConfigs{Name: name, Logger: logger},
+		BaseConfigs:          service.BaseConfigs{Name: name, Logger: logger},
 		EnableSignalHandling: true,
 		TelemetryCreate:      true,
 		TelemetryAddress:     cfg.AdvancerTelemetryAddress,

internal/evmreader/evmreader.go

@@ -110,6 +110,9 @@ func (r *Service) Tick(ctx context.Context) (bool, error) {
 		return false, err
 	}
 
+	now := time.Now()
+	r.lastSuccessfulPoll.Store(&now)
+
 	if blockNumber != r.lastBlockNumber.Load() {
 		r.lastBlockNumber.Store(blockNumber)
 		r.Logger.Info("Got new block header", "block", blockNumber, "policy", r.defaultBlock)
@@ -122,6 +125,10 @@ func (r *Service) Tick(ctx context.Context) (bool, error) {
 	return false, nil
 }
 
+func (r *Service) Ready() bool {
+	return time.Since(*r.lastSuccessfulPoll.Load()) < r.pollingMaxWait
+}
+
 func (r *Service) processBlockHead(
 	ctx context.Context,
 	blockNumber uint64,

internal/evmreader/evmreader_test.go

@@ -55,31 +55,26 @@ func (s *EvmReaderSuite) SetupTest() {
 	s.inputBox = newMockInputBox().SetupDefaultBehavior()
 	s.contractFactory = newMockAdapterFactory().SetupDefaultBehavior(s.applicationContract1, s.applicationContract2, s.inputBox)
 
-	s.evmReader = &Service{
-		client:             s.client,
-		repository:         s.repository,
-		defaultBlock:       DefaultBlock_Latest,
-		inputReaderEnabled: true,
-		hasEnabledApps:     true,
-		adapterFactory:     s.contractFactory,
-	}
-
 	logLevel, err := config.GetLogLevel()
 	s.Require().NoError(err)
 
-	serviceArgs := &service.TickServiceConfigs{
-		BaseConfigs: service.BaseConfigs{
-			Name:     "evm-reader",
-			LogLevel: logLevel,
+	evmReader, err := Create(s.T().Context(), &CreateInfo{
+		Config: config.EvmreaderConfig{
+			LogLevel:                   logLevel,
+			BlockchainHttpRetryMaxWait: 200 * time.Millisecond,
+			BlockchainDefaultBlock:     DefaultBlock_Latest,
+			FeatureInputReaderEnabled:  true,
+			EvmReaderPollingInterval:   100 * time.Millisecond,
 		},
-		PollInterval: 100 * time.Millisecond,
-	}
-	err = service.InitTickServiceTemplate(&s.evmReader.TickServiceTemplate, serviceArgs, s.evmReader)
+		Repository:     s.repository,
+		EthClient:      s.client,
+		AdapterFactory: s.contractFactory,
+	})
 	s.Require().NoError(err)
-
-	s.evmReader.resolver = newApplicationAdapterResolver(s.evmReader.Logger, s.contractFactory)
+	s.evmReader = evmReader.(*Service)
 
 	supCfg := service.SupervisorConfigs{
+		BaseConfigs: service.BaseConfigs{LogLevel: logLevel},
 		Factories: []service.FactoryFunction{
 			func(context.Context, *service.Supervisor) (service.SupervisedService, error) {
 				return s.evmReader, nil
@@ -132,68 +127,101 @@ func wasntNotified(ch <-chan struct{}) bool {
 }
 
 // Service tests
-func (s *EvmReaderSuite) TestItStopsWhenSupervisorIsStoppedAfterFirstHeader() {
+func (s *EvmReaderSuite) TestItStopsWhenContextIsCanceled() {
 	called := newCallNotification(s.client.EnqueueNewHead(100))
 
+	ctx, cancel := context.WithCancel(s.T().Context())
+
 	done := make(chan struct{})
 	go func() {
-		err := s.supervisor.Serve(s.T().Context())
-		s.Require().NoError(err)
+		err := s.evmReader.Serve(ctx)
+		s.Require().ErrorIs(err, context.Canceled)
 		close(done)
 	}()
 
 	s.Require().True(waitNotification(called), "evmreader did not read new header")
 
-	s.supervisor.Stop(false)
+	cancel()
 
 	s.Require().True(waitNotification(done), "evmreader did not stop after context cancelation")
 }
 
 func (s *EvmReaderSuite) TestReadyReflectsServeLifecycle() {
 	called := newCallNotification(s.client.EnqueueNewHead(100))
 
-	s.Require().False(s.supervisor.Ready())
+	s.Require().False(s.evmReader.Ready())
+
+	ctx, cancel := context.WithCancel(s.T().Context())
 
 	done := make(chan struct{})
 	go func() {
-		err := s.supervisor.Serve(s.T().Context())
-		s.Require().NoError(err)
+		err := s.evmReader.Serve(ctx)
+		s.Require().ErrorIs(err, context.Canceled)
 		close(done)
 	}()
 
 	s.Require().True(waitNotification(called))
-	s.Require().True(s.supervisor.Ready())
-	s.Require().True(wasntNotified(done))
+	s.Require().True(s.evmReader.Ready())
 
-	s.supervisor.Stop(false)
+	s.Require().True(wasntNotified(done))
+	cancel()
 	s.Require().True(waitNotification(done))
-	s.Require().False(s.supervisor.Ready())
+
+	s.Require().True(s.evmReader.Ready())
+	time.Sleep(s.evmReader.pollingMaxWait)
+	s.Require().False(s.evmReader.Ready())
 }
 
-func (s *EvmReaderSuite) TestReadyDoesNotDependOnPollingSuccess() {
+func (s *EvmReaderSuite) TestNotReadyWhilePollingFails() {
 	var hdr *types.Header
 	called := newCallNotification(s.client.On("HeaderByNumber",
 		mock.Anything,
 		mock.Anything,
+	).Return(hdr, errors.New("transient connection error")))
+
+	s.Require().False(s.evmReader.Ready())
+
+	go func() {
+		err := s.evmReader.Serve(s.T().Context())
+		s.Require().ErrorIs(err, context.Canceled)
+	}()
+
+	s.Require().True(waitNotification(called))
+	s.Require().False(s.evmReader.Ready())
+}
+
+func (s *EvmReaderSuite) TestReadyIsRecoveredOnPollingSuccess() {
+	var hdr *types.Header
+	failCalled := newCallNotification(s.client.On("HeaderByNumber",
+		mock.Anything,
+		mock.Anything,
 	).Return(hdr, errors.New("transient connection error")).Once())
+	called := newCallNotification(s.client.EnqueueNewHead(100))
 
-	s.Require().False(s.supervisor.Ready())
+	s.Require().False(s.evmReader.Ready())
+
+	ctx, cancel := context.WithCancel(s.T().Context())
 
 	done := make(chan struct{})
 	go func() {
-		err := s.supervisor.Serve(s.T().Context())
-		s.Require().NoError(err)
+		err := s.evmReader.Serve(ctx)
+		s.Require().ErrorIs(err, context.Canceled)
 		close(done)
 	}()
 
-	s.Require().True(waitNotification(called))
-	s.Require().True(s.supervisor.Ready())
-	s.Require().True(wasntNotified(done))
+	s.Require().True(waitNotification(failCalled))
+	s.Require().False(s.evmReader.Ready())
 
-	s.supervisor.Stop(false)
+	s.Require().True(waitNotification(called))
+	s.Require().True(s.evmReader.Ready())
 
+	s.Require().True(wasntNotified(done))
+	cancel()
 	s.Require().True(waitNotification(done))
-	s.Require().False(s.supervisor.Ready())
+
+	s.Require().True(s.evmReader.Ready())
+	time.Sleep(s.evmReader.pollingMaxWait)
+	s.Require().False(s.evmReader.Ready())
 }
 
 func (s *EvmReaderSuite) TestTickScansWithServiceContext() {

internal/evmreader/mocks_test.go

@@ -76,6 +76,7 @@ func newMockEthClient() *MockEthClient {
 }
 
 func (m *MockEthClient) SetupDefaultBehavior() *MockEthClient {
+	m.On("ChainID", mock.Anything).Return(big.NewInt(0), nil)
 	return m
 }
 
@@ -189,6 +190,10 @@ func newMockRepository() *MockRepository {
 }
 
 func (m *MockRepository) SetupDefaultBehavior() *MockRepository {
+	m.On("LoadNodeConfigRaw", mock.Anything, EvmReaderConfigKey).
+		Return(([]byte)(nil), time.Time{}, time.Time{}, repository.ErrNotFound)
+	m.On("SaveNodeConfigRaw", mock.Anything, EvmReaderConfigKey, mock.Anything).
+		Return(nil)
 
 	apps := copyApplications(applications)
 	m.On("ListApplications",

internal/evmreader/service.go

@@ -10,6 +10,7 @@ import (
 	"log/slog"
 	"math/big"
 	"sync/atomic"
+	"time"
 
 	"github.com/cartesi/rollups-node/internal/config"
 	. "github.com/cartesi/rollups-node/internal/model"
@@ -20,10 +21,11 @@ import (
 )
 
 type CreateInfo struct {
-	Config     config.EvmreaderConfig
-	Logger     *slog.Logger
-	EthClient  *ethclient.Client
-	Repository EvmReaderRepository
+	Config         config.EvmreaderConfig
+	Logger         *slog.Logger
+	Repository     EvmReaderRepository
+	EthClient      EthClientInterface
+	AdapterFactory AdapterFactory
 }
 
 type Service struct {
@@ -38,6 +40,8 @@ type Service struct {
 	hasEnabledApps     bool
 	inputReaderEnabled bool
 	lastBlockNumber    atomic.Uint64
+	lastSuccessfulPoll atomic.Pointer[time.Time]
+	pollingMaxWait     time.Duration
 }
 
 const EvmReaderConfigKey = "evm-reader"
@@ -116,15 +120,27 @@ func Create(ctx context.Context, c *CreateInfo) (service.SupervisedService, erro
 	s.defaultBlock = nodeConfig.DefaultBlock
 	s.inputReaderEnabled = nodeConfig.InputReaderEnabled
 	s.hasEnabledApps = true
-	s.adapterFactory = &DefaultAdapterFactory{
-		Client: ethClient,
-		Filter: ethutil.Filter{
-			MinChunkSize: ethutil.DefaultMinChunkSize,
-			MaxChunkSize: new(big.Int).SetUint64(c.Config.BlockchainMaxBlockRange),
-			Logger:       s.Logger,
-		},
+
+	if c.AdapterFactory != nil {
+		s.adapterFactory = c.AdapterFactory
+	} else {
+		fullEthClient, ok := ethClient.(*ethclient.Client)
+		if !ok {
+			return nil, fmt.Errorf("EthClient must be *ethclient.Client when AdapterFactory is not provided")
+		}
+		s.adapterFactory = &DefaultAdapterFactory{
+			Client: fullEthClient,
+			Filter: ethutil.Filter{
+				MinChunkSize: ethutil.DefaultMinChunkSize,
+				MaxChunkSize: new(big.Int).SetUint64(c.Config.BlockchainMaxBlockRange),
+				Logger:       s.Logger,
+			},
+		}
 	}
+
 	s.resolver = newApplicationAdapterResolver(s.Logger, s.adapterFactory)
+	s.pollingMaxWait = c.Config.BlockchainHttpRetryMaxWait
+	s.lastSuccessfulPoll.Store(&time.Time{})
 
 	s.Logger.Info("Created", "config", c.Config)
 

internal/prt/service.go

@@ -21,9 +21,9 @@ import (
 )
 
 type CreateInfo struct {
-	Config         config.PrtConfig
-	Logger         *slog.Logger
-	Repository     repository.Repository
+	Config     config.PrtConfig
+	Logger     *slog.Logger
+	Repository repository.Repository
 }
 
 type Service struct {

pkg/service/supervisor_test.go

@@ -367,7 +367,7 @@ func (s *SupervisorSuite) TestFailedInitialization() {
 	errorChild := newErrorService("error-child")
 
 	supervisor := NewSupervisor(&SupervisorConfigs{
-		BaseConfigs: BaseConfigs{Name:   s.T().Name()},
+		BaseConfigs: BaseConfigs{Name: s.T().Name()},
 		Factories: []FactoryFunction{
 			func(context.Context, *Supervisor) (SupervisedService, error) {
 				return healthyChild, nil