fix(syncer) - Metadata syncer bug fixes/improvements (#2374)

* Debug logs

* Remove sleeping when no metadata was refreshed

* Update comment

* Add condition not to stream when no validators fetched

* Syncer - change batch fetch condition

* Share Storage - update 'BeaconMetadataLastUpdated' for unchanged shares

* Implement unit test, fix one of the conditions

* SyncAll - unconditionally sync all shares

* Add comment and condition for early return

* Update comment

* Populate metadata also for keys without CL provided metadata

* fix/add more tests, linter

* Minor refactor, added comment

* Change some log levels to debug

* Do not persist shares with unchanged metadata

* fmt, add comments

Author: oleg-ssvlabs

cli/operator/node.go

@@ -582,7 +582,7 @@ var StartNodeCmd = &cobra.Command{
 			nodeProber.AddNode("event syncer", eventSyncer)
 		}
 
-		if _, err := metadataSyncer.SyncOnStartup(cmd.Context()); err != nil {
+		if _, err := metadataSyncer.SyncAll(cmd.Context()); err != nil {
 			logger.Fatal("failed to sync metadata on startup", zap.Error(err))
 		}
 

logging/names.go

@@ -1,18 +1,19 @@
 package logging
 
 const (
-	NameBootNode         = "BootNode"
-	NameController       = "Controller"
-	NameDiscoveryService = "DiscoveryService"
-	NameDutyScheduler    = "DutyScheduler"
-	NameEthClient        = "EthClient"
-	NameMetricsHandler   = "MetricsHandler"
-	NameOperator         = "Operator"
-	NameP2PNetwork       = "P2PNetwork"
-	NameSignerStorage    = "SignerStorage"
-	NameValidator        = "Validator"
-	NameWSServer         = "WSServer"
-	NameConnHandler      = "ConnHandler"
+	NameBootNode            = "BootNode"
+	NameController          = "Controller"
+	NameDiscoveryService    = "DiscoveryService"
+	NameDutyScheduler       = "DutyScheduler"
+	NameEthClient           = "EthClient"
+	NameMetricsHandler      = "MetricsHandler"
+	NameOperator            = "Operator"
+	NameP2PNetwork          = "P2PNetwork"
+	NameSignerStorage       = "SignerStorage"
+	NameValidator           = "Validator"
+	NameWSServer            = "WSServer"
+	NameConnHandler         = "ConnHandler"
+	NameShareMetadataSyncer = "ShareMetadataSyncer"
 
 	NameBadgerDBLog       = "BadgerDBLog"
 	NameBadgerDBReporting = "BadgerDBReporting"

operator/validator/metadata/syncer.go

@@ -13,6 +13,7 @@ import (
 
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 
+	"github.com/ssvlabs/ssv/logging"
 	"github.com/ssvlabs/ssv/logging/fields"
 	networkcommons "github.com/ssvlabs/ssv/network/commons"
 	"github.com/ssvlabs/ssv/protocol/v2/blockchain/beacon"
@@ -27,7 +28,19 @@ const (
 	defaultSyncInterval      = 12 * time.Minute
 	defaultStreamInterval    = 2 * time.Second
 	defaultUpdateSendTimeout = 30 * time.Second
-	batchSize                = 512
+	// NOTE:
+	// A higher value of `batchSize` results in fewer HTTP calls to the Consensus Node,
+	// but each call will have larger payloads and responses. While this speeds up
+	// metadata synchronization, it also increases the risk of timeouts.
+	//
+	// The value of `batchSize` should depend on the number of validators the node handles
+	// (especially relevant when comparing Exporter vs. Non-Exporter nodes)
+	// and the sync interval (how often metadata should be refreshed).
+	//
+	// ⚠️ Caution: Since there is no prioritization implemented, if the node cannot
+	// sync all validator shares within the given sync interval, there is a high risk
+	// that some validators will not be refreshed for an unpredictable amount of time.
+	batchSize = 512
 )
 
 type Syncer struct {
@@ -60,7 +73,7 @@ func NewSyncer(
 	opts ...Option,
 ) *Syncer {
 	u := &Syncer{
-		logger:            logger,
+		logger:            logger.Named(logging.NameShareMetadataSyncer),
 		shareStorage:      shareStorage,
 		validatorStore:    validatorStore,
 		beaconNode:        beaconNode,
@@ -85,7 +98,10 @@ func WithSyncInterval(interval time.Duration) Option {
 	}
 }
 
-func (s *Syncer) SyncOnStartup(ctx context.Context) (beacon.ValidatorMetadataMap, error) {
+// SyncAll loads all non-liquidated validator shares that belong to operator's subnets,
+// and triggers a full metadata synchronization for them.
+// It returns a mapping of validator public keys to their updated metadata.
+func (s *Syncer) SyncAll(ctx context.Context) (beacon.ValidatorMetadataMap, error) {
 	subnetsBuf := new(big.Int)
 	ownSubnets := s.selfSubnets(subnetsBuf)
 
@@ -100,19 +116,9 @@ func (s *Syncer) SyncOnStartup(ctx context.Context) (beacon.ValidatorMetadataMap
 		return nil, nil
 	}
 
-	// Skip syncing if metadata was already fetched before
-	// to prevent blocking startup after first sync.
-	needToSync := false
 	pubKeysToFetch := make([]spectypes.ValidatorPK, 0, len(shares))
 	for _, share := range shares {
 		pubKeysToFetch = append(pubKeysToFetch, share.ValidatorPubKey)
-		if !share.HasBeaconMetadata() {
-			needToSync = true
-		}
-	}
-	if !needToSync {
-		// Stream should take it over from here.
-		return nil, nil
 	}
 
 	// Sync all pubkeys that belong to own subnets. We don't need to batch them because we need to wait here until all of them are synced.
@@ -122,8 +128,12 @@ func (s *Syncer) SyncOnStartup(ctx context.Context) (beacon.ValidatorMetadataMap
 // Sync retrieves metadata for the provided public keys and updates storage accordingly.
 // Returns updated metadata for keys that had changes. Returns nil if no keys were provided or no updates occurred.
 func (s *Syncer) Sync(ctx context.Context, pubKeys []spectypes.ValidatorPK) (beacon.ValidatorMetadataMap, error) {
+	if len(pubKeys) == 0 {
+		return nil, nil
+	}
+
 	fetchStart := time.Now()
-	metadata, err := s.Fetch(ctx, pubKeys)
+	metadata, err := s.fetchMetadata(ctx, pubKeys)
 	if err != nil {
 		return nil, fmt.Errorf("fetch metadata: %w", err)
 	}
@@ -150,11 +160,10 @@ func (s *Syncer) Sync(ctx context.Context, pubKeys []spectypes.ValidatorPK) (bea
 	return updatedValidators, nil
 }
 
-func (s *Syncer) Fetch(ctx context.Context, pubKeys []spectypes.ValidatorPK) (beacon.ValidatorMetadataMap, error) {
-	if len(pubKeys) == 0 {
-		return nil, nil
-	}
-
+// fetchMetadata is responsible for fetching validator metadata from the beacon node for the provided public keys.
+// The beacon node response is sometimes empty for certain public keys — for such validators,
+// the ValidatorMetadataMap will contain empty metadata objects.
+func (s *Syncer) fetchMetadata(ctx context.Context, pubKeys []spectypes.ValidatorPK) (beacon.ValidatorMetadataMap, error) {
 	blsPubKeys := make([]phase0.BLSPubKey, len(pubKeys))
 	for i, pk := range pubKeys {
 		blsPubKeys[i] = phase0.BLSPubKey(pk)
@@ -165,7 +174,12 @@ func (s *Syncer) Fetch(ctx context.Context, pubKeys []spectypes.ValidatorPK) (be
 		return nil, fmt.Errorf("get validator data from beacon node: %w", err)
 	}
 
-	results := make(beacon.ValidatorMetadataMap, len(validatorsIndexMap))
+	results := make(beacon.ValidatorMetadataMap, len(pubKeys))
+
+	for _, key := range pubKeys {
+		results[key] = &beacon.ValidatorMetadata{}
+	}
+
 	for _, v := range validatorsIndexMap {
 		meta := &beacon.ValidatorMetadata{
 			Status:          v.Status,
@@ -179,6 +193,9 @@ func (s *Syncer) Fetch(ctx context.Context, pubKeys []spectypes.ValidatorPK) (be
 	return results, nil
 }
 
+// Stream continuously fetches and streams batches of validator metadata updates as they become available.
+// It yields updates through a channel (`SyncBatch`) and handles retries, sleeping between sync attempts
+// when all metadata is up to date. The loop respects the provided context and stops gracefully when canceled.
 func (s *Syncer) Stream(ctx context.Context) <-chan SyncBatch {
 	metadataUpdates := make(chan SyncBatch)
 
@@ -199,33 +216,33 @@ func (s *Syncer) Stream(ctx context.Context) <-chan SyncBatch {
 				continue
 			}
 
-			if len(batch.After) == 0 {
+			if len(batch.Before) == 0 {
+				s.logger.Debug("sleeping because all validators’ metadata has been refreshed.",
+					zap.Duration("sleep_for", s.streamInterval),
+					zap.Duration("refresh_interval", s.syncInterval))
 				if slept := s.sleep(ctx, s.streamInterval); !slept {
 					return
 				}
 				continue
 			}
 
-			// TODO: use time.After when Go is updated to 1.23
-			timer := time.NewTimer(s.updateSendTimeout)
 			select {
 			case metadataUpdates <- batch:
 				// Only sleep if there aren't more validators to fetch metadata for.
 				// It's done to wait for some data to appear. Without sleep, the next batch would likely be empty.
 				if done {
+					s.logger.Debug("sleeping after batch was streamed because all validators’ metadata has been refreshed.",
+						zap.Duration("sleep_for", s.streamInterval),
+						zap.Duration("refresh_interval", s.syncInterval))
 					if slept := s.sleep(ctx, s.streamInterval); !slept {
-						// canceled context
-						timer.Stop()
 						return
 					}
 				}
 			case <-ctx.Done():
-				timer.Stop()
 				return
-			case <-timer.C:
+			case <-time.After(s.updateSendTimeout):
 				s.logger.Warn("timed out waiting for sending update")
 			}
-			timer.Stop()
 		}
 	}()
 
@@ -289,7 +306,7 @@ func (s *Syncer) nextBatchFromDB(_ context.Context, subnetsBuf *big.Int) beacon.
 			staleShares = append(staleShares, share)
 		}
 
-		return len(newShares) < batchSize
+		return len(newShares)+len(staleShares) < batchSize
 	})
 
 	// Combine validators up to batchSize, prioritizing the new ones.
@@ -307,14 +324,10 @@ func (s *Syncer) nextBatchFromDB(_ context.Context, subnetsBuf *big.Int) beacon.
 }
 
 func (s *Syncer) sleep(ctx context.Context, d time.Duration) (slept bool) {
-	// TODO: use time.After when Go is updated to 1.23
-	timer := time.NewTimer(d)
-	defer timer.Stop()
-
 	select {
 	case <-ctx.Done():
 		return false
-	case <-timer.C:
+	case <-time.After(d):
 		return true
 	}
 }