use hooks to update fork value

beacon/goclient/dataversion.go

@@ -9,6 +9,8 @@ import (
 )
 
 func (gc *GoClient) DataVersion(epoch phase0.Epoch) spec.DataVersion {
+	gc.ForkLock.RLock()
+	defer gc.ForkLock.RUnlock()
 	if epoch < gc.ForkEpochAltair {
 		return spec.DataVersionPhase0
 	} else if epoch < gc.ForkEpochBellatrix {
@@ -23,72 +25,91 @@ func (gc *GoClient) DataVersion(epoch phase0.Epoch) spec.DataVersion {
 	return spec.DataVersionElectra
 }
 
-func fetchStaticValues(gc *GoClient) error {
-	// Fetch spec response
-	specResponse, err := gc.multiClient.Spec(gc.ctx, &api.SpecOpts{})
-	if err != nil {
-		return fmt.Errorf("failed to obtain spec response: %w", err)
-	}
+func (gc *GoClient) checkForkValues(specResponse *api.Response[map[string]any]) error {
+	// Validate the response.
 	if specResponse == nil {
 		return fmt.Errorf("spec response is nil")
 	}
 	if specResponse.Data == nil {
 		return fmt.Errorf("spec response data is nil")
 	}
 
-	// ALTAIR (required)
-	forkEpochRaw, ok := specResponse.Data["ALTAIR_FORK_EPOCH"]
-	if !ok {
-		return fmt.Errorf("altair fork epoch not known by chain")
-	}
-	forkEpoch, ok := forkEpochRaw.(uint64)
-	if !ok {
-		return fmt.Errorf("failed to decode altair fork epoch")
-	}
-	gc.ForkEpochAltair = phase0.Epoch(forkEpoch)
+	// Lock the fork values to ensure atomic read and update.
+	gc.ForkLock.Lock()
+	defer gc.ForkLock.Unlock()
 
-	// BELLATRIX (required)
-	forkEpochRaw, ok = specResponse.Data["BELLATRIX_FORK_EPOCH"]
-	if !ok {
-		return fmt.Errorf("bellatrix fork epoch not known by chain")
-	}
-	forkEpoch, ok = forkEpochRaw.(uint64)
-	if !ok {
-		return fmt.Errorf("failed to decode bellatrix fork epoch")
+	// We'll compute candidate new values first and update the fields only if all validations pass.
+	var newAltair, newBellatrix, newCapella, newDeneb, newElectra phase0.Epoch
+
+	// processFork is a helper to handle required forks.
+	// It retrieves the candidate fork epoch from the response,
+	// and compares it with the current stored value.
+	// If the candidate is greater than the current value, that's an error.
+	// Otherwise, it returns the lower value (or the candidate if the current value is zero).
+	processFork := func(forkName, key string, current phase0.Epoch) (phase0.Epoch, error) {
+		raw, ok := specResponse.Data[key]
+		if !ok {
+			return 0, fmt.Errorf("%s fork epoch not known by chain", forkName)
+		}
+		forkVal, ok := raw.(uint64)
+		if !ok {
+			return 0, fmt.Errorf("failed to decode %s fork epoch", forkName)
+		}
+		candidate := phase0.Epoch(forkVal)
+		// Error if the candidate is greater than the current (non-zero) value.
+		if current != 0 && candidate > current {
+			return 0, fmt.Errorf("new %s fork epoch (%d) is greater than current (%d)", forkName, candidate, current)
+		}
+		// Otherwise, if current is zero or the candidate is lower, use the candidate.
+		if current == 0 || candidate < current {
+			return candidate, nil
+		}
+		// If they are equal, no update is necessary.
+		return current, nil
 	}
-	gc.ForkEpochBellatrix = phase0.Epoch(forkEpoch)
 
-	// CAPELLA (required)
-	forkEpochRaw, ok = specResponse.Data["CAPELLA_FORK_EPOCH"]
-	if !ok {
-		return fmt.Errorf("capella fork epoch not known by chain")
+	var err error
+	// Process required forks.
+	if newAltair, err = processFork("ALTAIR", "ALTAIR_FORK_EPOCH", gc.ForkEpochAltair); err != nil {
+		return err
 	}
-	forkEpoch, ok = forkEpochRaw.(uint64)
-	if !ok {
-		return fmt.Errorf("failed to decode capella fork epoch")
+	if newBellatrix, err = processFork("BELLATRIX", "BELLATRIX_FORK_EPOCH", gc.ForkEpochBellatrix); err != nil {
+		return err
 	}
-	gc.ForkEpochCapella = phase0.Epoch(forkEpoch)
-
-	// DENEB (required)
-	forkEpochRaw, ok = specResponse.Data["DENEB_FORK_EPOCH"]
-	if !ok {
-		return fmt.Errorf("deneb fork epoch not known by chain")
+	if newCapella, err = processFork("CAPELLA", "CAPELLA_FORK_EPOCH", gc.ForkEpochCapella); err != nil {
+		return err
 	}
-	forkEpoch, ok = forkEpochRaw.(uint64)
-	if !ok {
-		return fmt.Errorf("failed to decode deneb fork epoch")
+	if newDeneb, err = processFork("DENEB", "DENEB_FORK_EPOCH", gc.ForkEpochDeneb); err != nil {
+		return err
 	}
-	gc.ForkEpochDeneb = phase0.Epoch(forkEpoch)
 
-	// ELECTRA (optional; set only if found)
-	forkEpochRaw, ok = specResponse.Data["ELECTRA_FORK_EPOCH"]
-	if ok {
-		forkEpoch, ok := forkEpochRaw.(uint64)
+	// Process the optional ELECTRA fork.
+	// If the key exists, perform the same validation; otherwise, keep the current value.
+	if raw, ok := specResponse.Data["ELECTRA_FORK_EPOCH"]; ok {
+		forkVal, ok := raw.(uint64)
 		if !ok {
-			return fmt.Errorf("failed to decode electra fork epoch")
+			return fmt.Errorf("failed to decode ELECTRA fork epoch")
 		}
-		gc.ForkEpochElectra = phase0.Epoch(forkEpoch)
+		candidate := phase0.Epoch(forkVal)
+		if gc.ForkEpochElectra != 0 && candidate > gc.ForkEpochElectra {
+			return fmt.Errorf("new ELECTRA fork epoch (%d) is greater than current (%d)", candidate, gc.ForkEpochElectra)
+		}
+		if gc.ForkEpochElectra == 0 || candidate < gc.ForkEpochElectra {
+			newElectra = candidate
+		} else {
+			newElectra = gc.ForkEpochElectra
+		}
+	} else {
+		newElectra = gc.ForkEpochElectra
 	}
 
+	// At this point, no error was encountered.
+	// Update all fork values atomically.
+	gc.ForkEpochAltair = newAltair
+	gc.ForkEpochBellatrix = newBellatrix
+	gc.ForkEpochCapella = newCapella
+	gc.ForkEpochDeneb = newDeneb
+	gc.ForkEpochElectra = newElectra
+
 	return nil
 }

beacon/goclient/goclient.go

@@ -141,6 +141,7 @@ type GoClient struct {
 	commonTimeout time.Duration
 	longTimeout   time.Duration
 
+	ForkLock           sync.RWMutex
 	ForkEpochElectra   phase0.Epoch
 	ForkEpochDeneb     phase0.Epoch
 	ForkEpochCapella   phase0.Epoch
@@ -205,11 +206,6 @@ func New(
 
 	client.nodeSyncingFn = client.nodeSyncing
 
-	// Get the fork epochs.
-	if err := fetchStaticValues(client); err != nil {
-		return nil, fmt.Errorf("fetch static values: %w", err)
-	}
-
 	go client.registrationSubmitter(slotTickerProvider)
 	// Start automatic expired item deletion for attestationDataCache.
 	go client.attestationDataCache.Start()
@@ -304,6 +300,24 @@ func (gc *GoClient) singleClientHooks() *eth2clienthttp.Hooks {
 				)
 				return // Tests may override Fatal's behavior
 			}
+
+			spec, err := s.Spec(ctx, &api.SpecOpts{})
+			if err != nil {
+				gc.log.Error(clResponseErrMsg,
+					zap.String("address", s.Address()),
+					zap.String("api", "Spec"),
+					zap.Error(err),
+				)
+				return
+			}
+
+			if err := gc.checkForkValues(spec); err != nil {
+				gc.log.Error("failed to check fork values",
+					zap.String("address", s.Address()),
+					zap.Error(err),
+				)
+				return
+			}
 		},
 		OnInactive: func(ctx context.Context, s *eth2clienthttp.Service) {
 			gc.log.Warn("consensus client disconnected",