x

Author: alpe

block/internal/syncing/syncer.go

@@ -151,14 +151,28 @@ func (s *Syncer) Start(ctx context.Context) error {
 
 // Stop shuts down the syncing component
 func (s *Syncer) Stop() error {
-	if s.cancel != nil {
-		s.cancel()
+	if s.cancel == nil {
+		return nil
 	}
+	s.cancel()
 	s.wg.Wait()
 	s.logger.Info().Msg("syncer stopped")
+	close(s.heightInCh)
+	s.cancel = nil
 	return nil
 }
 
+var alwaysTickCh chan time.Time
+
+func init() {
+	alwaysTickCh = make(chan time.Time)
+	close(alwaysTickCh) // always picked in select
+}
+
+func (s *Syncer) HasUnprocessedEvents() bool {
+	return len(s.heightInCh) != 0 || s.tryFetchFromP2P(&s.lastState.LastBlockHeight, &s.lastState.LastBlockHeight, alwaysTickCh)
+}
+
 // GetLastState returns the current state
 func (s *Syncer) GetLastState() types.State {
 	s.lastStateMtx.RLock()
@@ -222,8 +236,10 @@ func (s *Syncer) processLoop() {
 		select {
 		case <-s.ctx.Done():
 			return
-		case heightEvent := <-s.heightInCh:
-			s.processHeightEvent(&heightEvent)
+		case heightEvent, ok := <-s.heightInCh:
+			if ok {
+				s.processHeightEvent(&heightEvent)
+			}
 		}
 	}
 }
@@ -409,8 +425,7 @@ func (s *Syncer) processHeightEvent(event *common.DAHeightEvent) {
 	}
 
 	// If this is not the next block in sequence, store as pending event
-	// This check is crucial as try
-	//SyncNextBlock simply attempts to sync the next block
+	// This check is crucial as trySyncNextBlock simply attempts to sync the next block
 	if height != currentHeight+1 {
 		s.cache.SetPendingEvent(height, event)
 		s.logger.Debug().Uint64("height", height).Uint64("current_height", currentHeight).Msg("stored as pending event")

node/failover.go

@@ -0,0 +1,229 @@
+package node
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/evstack/ev-node/block"
+	coreda "github.com/evstack/ev-node/core/da"
+	coreexecutor "github.com/evstack/ev-node/core/execution"
+	coresequencer "github.com/evstack/ev-node/core/sequencer"
+	"github.com/evstack/ev-node/pkg/config"
+	genesispkg "github.com/evstack/ev-node/pkg/genesis"
+	"github.com/evstack/ev-node/pkg/p2p"
+	"github.com/evstack/ev-node/pkg/p2p/key"
+	rpcserver "github.com/evstack/ev-node/pkg/rpc/server"
+	"github.com/evstack/ev-node/pkg/signer"
+	"github.com/evstack/ev-node/pkg/store"
+	evsync "github.com/evstack/ev-node/pkg/sync"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/rs/zerolog"
+)
+
+// failoverState collect the components to reset when switching modes.
+type failoverState struct {
+	logger zerolog.Logger
+
+	p2pClient         *p2p.Client
+	headerSyncService *evsync.HeaderSyncService
+	dataSyncService   *evsync.DataSyncService
+	rpcServer         *http.Server
+	bc                *block.Components
+}
+
+func newSyncMode(
+	nodeConfig config.Config,
+	nodeKey *key.NodeKey,
+	genesis genesispkg.Genesis,
+	database ds.Batching,
+	exec coreexecutor.Executor,
+	da coreda.DA,
+	logger zerolog.Logger,
+	rktStore store.Store,
+	mainKV ds.Batching,
+	blockMetrics *block.Metrics,
+	nodeOpts NodeOptions,
+	raftNode block.RaftNode,
+) (*failoverState, error) {
+	blockComponentsFn := func(headerSyncService *evsync.HeaderSyncService, dataSyncService *evsync.DataSyncService) (*block.Components, error) {
+		return block.NewSyncComponents(
+			nodeConfig,
+			genesis,
+			rktStore,
+			exec,
+			da,
+			headerSyncService.Store(),
+			dataSyncService.Store(),
+			logger,
+			blockMetrics,
+			nodeOpts.BlockOptions,
+			raftNode,
+		)
+	}
+	return setupFailoverState(nodeConfig, nodeKey, database, genesis, logger, mainKV, rktStore, blockComponentsFn)
+}
+func newAggregatorMode(
+	nodeConfig config.Config,
+	nodeKey *key.NodeKey,
+	signer signer.Signer,
+	genesis genesispkg.Genesis,
+	database ds.Batching,
+	exec coreexecutor.Executor,
+	sequencer coresequencer.Sequencer,
+	da coreda.DA,
+	logger zerolog.Logger,
+	rktStore store.Store,
+	mainKV ds.Batching,
+	blockMetrics *block.Metrics,
+	nodeOpts NodeOptions,
+	raftNode block.RaftNode,
+) (*failoverState, error) {
+
+	blockComponentsFn := func(headerSyncService *evsync.HeaderSyncService, dataSyncService *evsync.DataSyncService) (*block.Components, error) {
+		return block.NewAggregatorComponents(
+			nodeConfig,
+			genesis,
+			rktStore,
+			exec,
+			sequencer,
+			da,
+			signer,
+			headerSyncService,
+			dataSyncService,
+			logger,
+			blockMetrics,
+			nodeOpts.BlockOptions,
+			raftNode,
+		)
+	}
+
+	return setupFailoverState(nodeConfig, nodeKey, database, genesis, logger, mainKV, rktStore, blockComponentsFn)
+}
+
+func setupFailoverState(
+	nodeConfig config.Config,
+	nodeKey *key.NodeKey,
+	database ds.Batching,
+	genesis genesispkg.Genesis,
+	logger zerolog.Logger,
+	mainKV ds.Batching,
+	rktStore store.Store,
+	buildComponentsFn func(headerSyncService *evsync.HeaderSyncService, dataSyncService *evsync.DataSyncService) (*block.Components, error),
+) (*failoverState, error) {
+	p2pClient, err := p2p.NewClient(nodeConfig.P2P, nodeKey.PrivKey, database, genesis.ChainID, logger, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	headerSyncService, err := evsync.NewHeaderSyncService(mainKV, nodeConfig, genesis, p2pClient, logger.With().Str("component", "HeaderSyncService").Logger())
+	if err != nil {
+		return nil, fmt.Errorf("error while initializing HeaderSyncService: %w", err)
+	}
+
+	dataSyncService, err := evsync.NewDataSyncService(mainKV, nodeConfig, genesis, p2pClient, logger.With().Str("component", "DataSyncService").Logger())
+	if err != nil {
+		return nil, fmt.Errorf("error while initializing DataSyncService: %w", err)
+	}
+
+	bestKnownHeightProvider := func() uint64 {
+		hHeight := headerSyncService.Store().Height()
+		dHeight := dataSyncService.Store().Height()
+		return min(hHeight, dHeight)
+	}
+	handler, err := rpcserver.NewServiceHandler(rktStore, p2pClient, genesis.ProposerAddress, logger, nodeConfig, bestKnownHeightProvider)
+	if err != nil {
+		return nil, fmt.Errorf("error creating RPC handler: %w", err)
+	}
+
+	rpcServer := &http.Server{
+		Addr:         nodeConfig.RPC.Address,
+		Handler:      handler,
+		ReadTimeout:  10 * time.Second,
+		WriteTimeout: 10 * time.Second,
+		IdleTimeout:  120 * time.Second,
+	}
+	bc, err := buildComponentsFn(headerSyncService, dataSyncService)
+	if err != nil {
+		return nil, fmt.Errorf("build follower components: %w", err)
+	}
+
+	return &failoverState{
+		logger:            logger,
+		p2pClient:         p2pClient,
+		headerSyncService: headerSyncService,
+		dataSyncService:   dataSyncService,
+		rpcServer:         rpcServer,
+		bc:                bc,
+	}, nil
+}
+
+func (f *failoverState) Run(ctx context.Context) (multiErr error) {
+	defer f.rpcServer.Shutdown(context.Background()) // nolint: errcheck
+
+	if err := f.p2pClient.Start(ctx); err != nil {
+		return err
+	}
+	defer f.p2pClient.Close() // nolint: errcheck
+
+	if err := f.headerSyncService.Start(ctx); err != nil {
+		return fmt.Errorf("error while starting header sync service: %w", err)
+	}
+	defer func() {
+		if err := f.headerSyncService.Stop(context.Background()); err != nil && !errors.Is(err, context.Canceled) {
+			multiErr = errors.Join(multiErr, fmt.Errorf("stopping header sync: %w", err))
+		}
+	}()
+
+	if err := f.dataSyncService.Start(ctx); err != nil {
+		return fmt.Errorf("error while starting data sync service: %w", err)
+	}
+	defer func() {
+		if err := f.dataSyncService.Stop(context.Background()); err != nil && !errors.Is(err, context.Canceled) {
+			multiErr = errors.Join(multiErr, fmt.Errorf("stopping data sync: %w", err))
+		}
+	}()
+
+	defer func() {
+		if f.bc.Syncer != nil {
+			for f.bc.Syncer.HasUnprocessedEvents() {
+				// give it some time to gracefully complete
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+		if err := f.bc.Stop(); err != nil && !errors.Is(err, context.Canceled) {
+			multiErr = errors.Join(multiErr, fmt.Errorf("stopping block components: %w", err))
+		}
+	}()
+
+	errChan := make(chan error)
+	go func() {
+		f.logger.Info().Str("addr", f.rpcServer.Addr).Msg("started RPC server")
+		if err := f.rpcServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			select {
+			case errChan <- fmt.Errorf("RPC server error: %w", err):
+			default:
+				f.logger.Error().Err(err).Msg("RPC server error")
+			}
+		}
+	}()
+
+	go func() {
+		if err := f.bc.Start(ctx); err != nil && !errors.Is(err, context.Canceled) {
+			select {
+			case errChan <- fmt.Errorf("components started with error: %w", err):
+			default:
+				f.logger.Error().Err(err).Msg("Components start error")
+			}
+			return
+		}
+		select {
+		case errChan <- nil:
+		default:
+		}
+	}()
+
+	return <-errChan
+}