sync2: add config validation (#6548)

## Motivation

Need to make sure syncv2 user-provided config is valid.

Author: ivan4th

node/node.go

@@ -807,7 +807,7 @@ func (app *App) initServices(ctx context.Context) error {
 		app.Config.Sync.MalSync.MinSyncPeers = max(1, app.Config.P2P.MinPeers)
 	}
 	app.syncLogger = app.addLogger(SyncLogger, lg)
-	syncer := syncer.NewSyncer(
+	syncer, err := syncer.NewSyncer(
 		app.cachedDB,
 		app.clock,
 		msh,
@@ -829,6 +829,9 @@ func (app *App) initServices(ctx context.Context) error {
 		syncer.WithConfig(syncerConf),
 		syncer.WithLogger(app.syncLogger.Zap()),
 	)
+	if err != nil {
+		return fmt.Errorf("create syncer: %w", err)
+	}
 	// TODO(dshulyak) this needs to be improved, but dependency graph is a bit complicated
 	beaconProtocol.SetSyncState(syncer)
 	hOracle.SetSync(syncer)

sync2/atxs.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"sync"
-	"time"
 
 	"github.com/jonboulle/clockwork"
 	"github.com/libp2p/go-libp2p/core/host"
@@ -32,40 +31,33 @@ const (
 )
 
 type ATXHandler struct {
-	logger           *zap.Logger
-	f                Fetcher
-	clock            clockwork.Clock
-	batchSize        int
-	maxAttempts      int
-	maxBatchRetries  int
-	failedBatchDelay time.Duration
+	logger *zap.Logger
+	f      Fetcher
+	clock  clockwork.Clock
+	cfg    Config
 }
 
 var _ multipeer.SyncKeyHandler = &ATXHandler{}
 
 func NewATXHandler(
 	logger *zap.Logger,
 	f Fetcher,
-	batchSize, maxAttempts, maxBatchRetries int,
-	failedBatchDelay time.Duration,
+	cfg Config,
 	clock clockwork.Clock,
 ) *ATXHandler {
 	if clock == nil {
 		clock = clockwork.NewRealClock()
 	}
 	return &ATXHandler{
-		f:                f,
-		logger:           logger,
-		clock:            clock,
-		batchSize:        batchSize,
-		maxAttempts:      maxAttempts,
-		maxBatchRetries:  maxBatchRetries,
-		failedBatchDelay: failedBatchDelay,
+		f:      f,
+		logger: logger,
+		clock:  clock,
+		cfg:    cfg,
 	}
 }
 
 type commitState struct {
-	state         map[types.ATXID]int
+	state         map[types.ATXID]uint
 	total         int
 	numDownloaded int
 	items         []types.ATXID
@@ -76,7 +68,7 @@ func (h *ATXHandler) setupState(
 	base rangesync.OrderedSet,
 	received rangesync.SeqResult,
 ) (*commitState, error) {
-	state := make(map[types.ATXID]int)
+	state := make(map[types.ATXID]uint)
 	for k := range received.Seq {
 		found, err := base.Has(k)
 		if err != nil {
@@ -95,15 +87,15 @@ func (h *ATXHandler) setupState(
 	return &commitState{
 		state: state,
 		total: len(state),
-		items: make([]types.ATXID, 0, h.batchSize),
+		items: make([]types.ATXID, 0, h.cfg.BatchSize),
 	}, nil
 }
 
 func (h *ATXHandler) getAtxs(ctx context.Context, cs *commitState) (bool, error) {
 	cs.items = cs.items[:0] // reuse the slice to reduce allocations
 	for id := range cs.state {
 		cs.items = append(cs.items, id)
-		if len(cs.items) == h.batchSize {
+		if uint(len(cs.items)) == h.cfg.BatchSize {
 			break
 		}
 	}
@@ -121,7 +113,7 @@ func (h *ATXHandler) getAtxs(ctx context.Context, cs *commitState) (bool, error)
 			h.logger.Debug("failed to download ATX",
 				zap.String("atx", id.ShortString()), zap.Error(err))
 			delete(cs.state, id)
-		case cs.state[id] >= h.maxAttempts-1:
+		case cs.state[id] >= h.cfg.MaxAttempts-1:
 			h.logger.Debug("failed to download ATX: max attempts reached",
 				zap.String("atx", id.ShortString()))
 			delete(cs.state, id)
@@ -145,7 +137,7 @@ func (h *ATXHandler) Commit(
 		return err
 	}
 	startTime := h.clock.Now()
-	batchAttemptsRemaining := h.maxBatchRetries
+	batchAttemptsRemaining := h.cfg.MaxBatchRetries
 	for len(cs.state) > 0 {
 		someSucceeded, err := h.getAtxs(ctx, cs)
 		batchErr := &fetch.BatchError{}
@@ -162,17 +154,17 @@ func (h *ATXHandler) Commit(
 			}
 			batchAttemptsRemaining--
 			h.logger.Debug("failed to download any ATXs: will retry batch",
-				zap.Int("remaining", batchAttemptsRemaining),
-				zap.Duration("delay", h.failedBatchDelay))
+				zap.Uint("remaining", batchAttemptsRemaining),
+				zap.Duration("delay", h.cfg.FailedBatchDelay))
 			select {
 			case <-ctx.Done():
 				return ctx.Err()
-			case <-h.clock.After(h.failedBatchDelay):
+			case <-h.clock.After(h.cfg.FailedBatchDelay):
 				continue
 			}
 		}
 
-		batchAttemptsRemaining = h.maxBatchRetries
+		batchAttemptsRemaining = h.cfg.MaxBatchRetries
 		elapsed := h.clock.Since(startTime)
 		h.logger.Debug("fetched atxs",
 			zap.Int("total", cs.total),
@@ -197,14 +189,17 @@ func NewMultiEpochATXSyncer(
 	hss HashSyncSource,
 	oldCfg, newCfg Config,
 	parallelLoadLimit int,
-) *MultiEpochATXSyncer {
+) (*MultiEpochATXSyncer, error) {
+	if !oldCfg.Validate(logger) || !newCfg.Validate(logger) {
+		return nil, errors.New("invalid config")
+	}
 	return &MultiEpochATXSyncer{
 		logger:            logger,
 		oldCfg:            oldCfg,
 		newCfg:            newCfg,
 		parallelLoadLimit: parallelLoadLimit,
 		hss:               hss,
-	}
+	}, nil
 }
 
 func (s *MultiEpochATXSyncer) load(newEpoch types.EpochID) error {
@@ -226,7 +221,10 @@ func (s *MultiEpochATXSyncer) load(newEpoch types.EpochID) error {
 			if epoch == newEpoch {
 				cfg = s.newCfg
 			}
-			hs := s.hss.CreateHashSync(name, cfg, epoch)
+			hs, err := s.hss.CreateHashSync(name, cfg, epoch)
+			if err != nil {
+				return fmt.Errorf("create ATX syncer for epoch %d: %w", epoch, err)
+			}
 			if err := hs.Load(); err != nil {
 				return fmt.Errorf("load ATX syncer for epoch %d: %w", epoch, err)
 			}
@@ -303,9 +301,9 @@ func NewATXSyncer(
 	peers *peers.Peers,
 	epoch types.EpochID,
 	enableActiveSync bool,
-) *P2PHashSync {
-	curSet := dbset.NewDBSet(db, atxsTable(epoch), 32, cfg.MaxDepth)
-	handler := NewATXHandler(logger, f, cfg.BatchSize, cfg.MaxAttempts, cfg.MaxBatchRetries, cfg.FailedBatchDelay, nil)
+) (*P2PHashSync, error) {
+	curSet := dbset.NewDBSet(db, atxsTable(epoch), 32, int(cfg.MaxDepth))
+	handler := NewATXHandler(logger, f, cfg, nil)
 	return NewP2PHashSync(logger, d, name, curSet, 32, peers, handler, cfg, enableActiveSync)
 }
 
@@ -338,6 +336,6 @@ func NewATXSyncSource(
 }
 
 // CreateHashSync implements HashSyncSource.
-func (as *ATXSyncSource) CreateHashSync(name string, cfg Config, epoch types.EpochID) HashSync {
+func (as *ATXSyncSource) CreateHashSync(name string, cfg Config, epoch types.EpochID) (HashSync, error) {
 	return NewATXSyncer(as.logger.Named(name), as.d, name, cfg, as.db, as.f, as.peers, epoch, as.enableActiveSync)
 }

sync2/atxs_test.go

@@ -36,13 +36,14 @@ func atxSeqResult(atxs []types.ATXID) rangesync.SeqResult {
 	}
 }
 
+var testCfg = sync2.Config{
+	BatchSize:        4,
+	MaxAttempts:      3,
+	MaxBatchRetries:  2,
+	FailedBatchDelay: 10 * time.Second,
+}
+
 func TestAtxHandler_Success(t *testing.T) {
-	const (
-		batchSize       = 4
-		maxAttempts     = 3
-		maxBatchRetries = 2
-		batchRetryDelay = 10 * time.Second
-	)
 	ctrl := gomock.NewController(t)
 	allAtxs := make([]types.ATXID, 10)
 	logger := zaptest.NewLogger(t)
@@ -52,7 +53,7 @@ func TestAtxHandler_Success(t *testing.T) {
 	}
 	f := NewMockFetcher(ctrl)
 	clock := clockwork.NewFakeClock()
-	h := sync2.NewATXHandler(logger, f, batchSize, maxAttempts, maxBatchRetries, batchRetryDelay, clock)
+	h := sync2.NewATXHandler(logger, f, testCfg, clock)
 	baseSet := mocks.NewMockOrderedSet(ctrl)
 	for _, id := range allAtxs {
 		baseSet.EXPECT().Has(rangesync.KeyBytes(id.Bytes()))
@@ -84,12 +85,6 @@ func TestAtxHandler_Success(t *testing.T) {
 }
 
 func TestAtxHandler_Retry(t *testing.T) {
-	const (
-		batchSize       = 4
-		maxAttempts     = 3
-		maxBatchRetries = 2
-		batchRetryDelay = 10 * time.Second
-	)
 	ctrl := gomock.NewController(t)
 	allAtxs := make([]types.ATXID, 10)
 	logger := zaptest.NewLogger(t)
@@ -99,7 +94,7 @@ func TestAtxHandler_Retry(t *testing.T) {
 	}
 	f := NewMockFetcher(ctrl)
 	clock := clockwork.NewFakeClock()
-	h := sync2.NewATXHandler(logger, f, batchSize, maxAttempts, maxBatchRetries, batchRetryDelay, clock)
+	h := sync2.NewATXHandler(logger, f, testCfg, clock)
 	baseSet := mocks.NewMockOrderedSet(ctrl)
 	for _, id := range allAtxs {
 		baseSet.EXPECT().Has(rangesync.KeyBytes(id.Bytes()))
@@ -159,7 +154,7 @@ func TestAtxHandler_Retry(t *testing.T) {
 			if ctx.Err() != nil {
 				return nil
 			}
-			clock.Advance(batchRetryDelay)
+			clock.Advance(testCfg.FailedBatchDelay)
 		}
 	})
 
@@ -170,19 +165,13 @@ func TestAtxHandler_Retry(t *testing.T) {
 }
 
 func TestAtxHandler_Cancel(t *testing.T) {
-	const (
-		batchSize       = 4
-		maxAttempts     = 3
-		maxBatchRetries = 2
-		batchRetryDelay = 10 * time.Second
-	)
 	atxID := types.RandomATXID()
 	ctrl := gomock.NewController(t)
 	logger := zaptest.NewLogger(t)
 	peer := p2p.Peer("foobar")
 	f := NewMockFetcher(ctrl)
 	clock := clockwork.NewFakeClock()
-	h := sync2.NewATXHandler(logger, f, batchSize, maxAttempts, maxBatchRetries, batchRetryDelay, clock)
+	h := sync2.NewATXHandler(logger, f, testCfg, clock)
 	baseSet := mocks.NewMockOrderedSet(ctrl)
 	baseSet.EXPECT().Has(rangesync.KeyBytes(atxID.Bytes())).Return(false, nil)
 	f.EXPECT().RegisterPeerHashes(peer, []types.Hash32{atxID.Hash32()})
@@ -200,12 +189,6 @@ func TestAtxHandler_Cancel(t *testing.T) {
 }
 
 func TestAtxHandler_BatchRetry(t *testing.T) {
-	const (
-		batchSize       = 4
-		maxAttempts     = 3
-		maxBatchRetries = 2
-		batchRetryDelay = 10 * time.Second
-	)
 	ctrl := gomock.NewController(t)
 	allAtxs := make([]types.ATXID, 10)
 	logger := zaptest.NewLogger(t)
@@ -215,7 +198,7 @@ func TestAtxHandler_BatchRetry(t *testing.T) {
 	}
 	clock := clockwork.NewFakeClock()
 	f := NewMockFetcher(ctrl)
-	h := sync2.NewATXHandler(logger, f, batchSize, maxAttempts, maxBatchRetries, batchRetryDelay, clock)
+	h := sync2.NewATXHandler(logger, f, testCfg, clock)
 	baseSet := mocks.NewMockOrderedSet(ctrl)
 	for _, id := range allAtxs {
 		baseSet.EXPECT().Has(rangesync.KeyBytes(id.Bytes()))
@@ -249,18 +232,12 @@ func TestAtxHandler_BatchRetry(t *testing.T) {
 			}
 			return nil
 		}).Times(3)
-	clock.Advance(batchRetryDelay)
+	clock.Advance(testCfg.FailedBatchDelay)
 	require.NoError(t, eg.Wait())
 	require.Empty(t, toFetch)
 }
 
 func TestAtxHandler_BatchRetry_Fail(t *testing.T) {
-	const (
-		batchSize       = 4
-		maxAttempts     = 3
-		maxBatchRetries = 2
-		batchRetryDelay = 10 * time.Second
-	)
 	ctrl := gomock.NewController(t)
 	allAtxs := make([]types.ATXID, 10)
 	logger := zaptest.NewLogger(t)
@@ -270,7 +247,7 @@ func TestAtxHandler_BatchRetry_Fail(t *testing.T) {
 	}
 	clock := clockwork.NewFakeClock()
 	f := NewMockFetcher(ctrl)
-	h := sync2.NewATXHandler(logger, f, batchSize, maxAttempts, maxBatchRetries, batchRetryDelay, clock)
+	h := sync2.NewATXHandler(logger, f, testCfg, clock)
 	baseSet := mocks.NewMockOrderedSet(ctrl)
 	for _, id := range allAtxs {
 		baseSet.EXPECT().Has(rangesync.KeyBytes(id.Bytes()))
@@ -296,7 +273,7 @@ func TestAtxHandler_BatchRetry_Fail(t *testing.T) {
 	})
 	for range 2 {
 		clock.BlockUntil(1)
-		clock.Advance(batchRetryDelay)
+		clock.Advance(testCfg.FailedBatchDelay)
 	}
 	require.Error(t, eg.Wait())
 }
@@ -309,7 +286,8 @@ func TestMultiEpochATXSyncer(t *testing.T) {
 	newCfg := sync2.DefaultConfig()
 	newCfg.MaxDepth = 24
 	hss := NewMockHashSyncSource(ctrl)
-	mhs := sync2.NewMultiEpochATXSyncer(logger, hss, oldCfg, newCfg, 1)
+	mhs, err := sync2.NewMultiEpochATXSyncer(logger, hss, oldCfg, newCfg, 1)
+	require.NoError(t, err)
 	ctx := context.Background()
 
 	lastSynced, err := mhs.EnsureSync(ctx, 0, 0)
@@ -319,7 +297,7 @@ func TestMultiEpochATXSyncer(t *testing.T) {
 	var syncActions []string
 	curIdx := 0
 	hss.EXPECT().CreateHashSync(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
-		func(name string, cfg sync2.Config, epoch types.EpochID) sync2.HashSync {
+		func(name string, cfg sync2.Config, epoch types.EpochID) (sync2.HashSync, error) {
 			idx := curIdx
 			curIdx++
 			syncActions = append(syncActions,
@@ -339,7 +317,7 @@ func TestMultiEpochATXSyncer(t *testing.T) {
 			hs.EXPECT().Stop().DoAndReturn(func() {
 				syncActions = append(syncActions, fmt.Sprintf("stop %d %s", idx, name))
 			}).AnyTimes()
-			return hs
+			return hs, nil
 		}).AnyTimes()
 
 	// Last wait epoch 3, new epoch 3

sync2/interface.go

@@ -23,7 +23,7 @@ type HashSync interface {
 }
 
 type HashSyncSource interface {
-	CreateHashSync(name string, cfg Config, epoch types.EpochID) HashSync
+	CreateHashSync(name string, cfg Config, epoch types.EpochID) (HashSync, error)
 }
 
 type LayerTicker interface {