merge commit

network/discovery/dv5_service.go

@@ -14,8 +14,6 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 
-	spectypes "github.com/ssvlabs/ssv-spec/types"
-
 	"github.com/ssvlabs/ssv/logging"
 	"github.com/ssvlabs/ssv/logging/fields"
 	"github.com/ssvlabs/ssv/network/commons"
@@ -62,10 +60,6 @@ type DiscV5Service struct {
 	subnets    []byte
 }
 
-func (dvs *DiscV5Service) UpdateDomainType(logger *zap.Logger, domain spectypes.DomainType) error {
-	return records.SetDomainTypeEntry(dvs.dv5Listener.LocalNode(), records.KeyDomainType, domain)
-}
-
 func newDiscV5Service(pctx context.Context, logger *zap.Logger, discOpts *Options) (Service, error) {
 	ctx, cancel := context.WithCancel(pctx)
 	dvs := DiscV5Service{

network/discovery/service.go

@@ -10,7 +10,6 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 	"go.uber.org/zap"
 
-	spectypes "github.com/ssvlabs/ssv-spec/types"
 	"github.com/ssvlabs/ssv/network/peers"
 	"github.com/ssvlabs/ssv/networkconfig"
 )
@@ -53,7 +52,6 @@ type Service interface {
 	RegisterSubnets(logger *zap.Logger, subnets ...int) error
 	DeregisterSubnets(logger *zap.Logger, subnets ...int) error
 	Bootstrap(logger *zap.Logger, handler HandleNewPeer) error
-	UpdateDomainType(logger *zap.Logger, domain spectypes.DomainType) error
 }
 
 // NewService creates new discovery.Service

network/network.go

@@ -6,7 +6,6 @@ import (
 
 	"go.uber.org/zap"
 
-	spectypes "github.com/ssvlabs/ssv-spec/types"
 	protocolp2p "github.com/ssvlabs/ssv/protocol/v2/p2p"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/queue"
 )
@@ -38,8 +37,6 @@ type P2PNetwork interface {
 	SubscribeAll(logger *zap.Logger) error
 	// SubscribeRandoms subscribes to random subnets
 	SubscribeRandoms(logger *zap.Logger, numSubnets int) error
-	// UpdateDomainType switches domain type at ENR when we reach fork epoch
-	UpdateDomainType(logger *zap.Logger, domain spectypes.DomainType) error
 	// UpdateScoreParams will update the scoring parameters of GossipSub
 	UpdateScoreParams(logger *zap.Logger)
 }

network/p2p/p2p.go

@@ -325,9 +325,10 @@ func (n *p2pNetwork) UpdateSubnets(logger *zap.Logger) {
 			continue
 		}
 
-		self := n.idx.Self()
-		self.Metadata.Subnets = records.Subnets(n.activeSubnets).String()
-		n.idx.UpdateSelfRecord(self)
+		n.idx.UpdateSelfRecord(func(self *records.NodeInfo) *records.NodeInfo {
+			self.Metadata.Subnets = records.Subnets(n.activeSubnets).String()
+			return self
+		})
 
 		var errs error
 		if len(addedSubnets) > 0 {
@@ -400,13 +401,3 @@ func (n *p2pNetwork) getMaxPeers(topic string) int {
 	}
 	return n.cfg.TopicMaxPeers
 }
-
-// UpdateDomainAtFork updates Domain Type at ENR node record after fork epoch.
-func (n *p2pNetwork) UpdateDomainType(logger *zap.Logger, domain spectypes.DomainType) error {
-	if err := n.disc.UpdateDomainType(logger, domain); err != nil {
-		logger.Error("could not update domain type", zap.Error(err))
-		return err
-	}
-	logger.Debug("updated and published ENR with domain type", fields.Domain(domain))
-	return nil
-}

network/peers/connections/handshaker.go

@@ -2,6 +2,7 @@ package connections
 
 import (
 	"context"
+	"encoding/hex"
 	"time"
 
 	libp2pnetwork "github.com/libp2p/go-libp2p/core/network"
@@ -14,6 +15,7 @@ import (
 	"github.com/ssvlabs/ssv/network/peers"
 	"github.com/ssvlabs/ssv/network/records"
 	"github.com/ssvlabs/ssv/network/streams"
+	"github.com/ssvlabs/ssv/networkconfig"
 	"github.com/ssvlabs/ssv/operator/keys"
 )
 
@@ -52,35 +54,38 @@ type handshaker struct {
 	ids        identify.IDService
 	net        libp2pnetwork.Network
 
-	subnetsProvider SubnetsProvider
+	domainTypeProvider networkconfig.DomainTypeProvider
+	subnetsProvider    SubnetsProvider
 }
 
 // HandshakerCfg is the configuration for creating an handshaker instance
 type HandshakerCfg struct {
-	Network         libp2pnetwork.Network
-	Streams         streams.StreamController
-	NodeInfos       peers.NodeInfoIndex
-	PeerInfos       peers.PeerInfoIndex
-	ConnIdx         peers.ConnectionIndex
-	SubnetsIdx      peers.SubnetsIndex
-	IDService       identify.IDService
-	OperatorSigner  keys.OperatorSigner
-	SubnetsProvider SubnetsProvider
+	Network            libp2pnetwork.Network
+	Streams            streams.StreamController
+	NodeInfos          peers.NodeInfoIndex
+	PeerInfos          peers.PeerInfoIndex
+	ConnIdx            peers.ConnectionIndex
+	SubnetsIdx         peers.SubnetsIndex
+	IDService          identify.IDService
+	OperatorSigner     keys.OperatorSigner
+	DomainTypeProvider networkconfig.DomainTypeProvider
+	SubnetsProvider    SubnetsProvider
 }
 
 // NewHandshaker creates a new instance of handshaker
 func NewHandshaker(ctx context.Context, cfg *HandshakerCfg, filters func() []HandshakeFilter) Handshaker {
 	h := &handshaker{
-		ctx:             ctx,
-		streams:         cfg.Streams,
-		nodeInfos:       cfg.NodeInfos,
-		connIdx:         cfg.ConnIdx,
-		subnetsIdx:      cfg.SubnetsIdx,
-		ids:             cfg.IDService,
-		filters:         filters,
-		peerInfos:       cfg.PeerInfos,
-		subnetsProvider: cfg.SubnetsProvider,
-		net:             cfg.Network,
+		ctx:                ctx,
+		streams:            cfg.Streams,
+		nodeInfos:          cfg.NodeInfos,
+		connIdx:            cfg.ConnIdx,
+		subnetsIdx:         cfg.SubnetsIdx,
+		ids:                cfg.IDService,
+		filters:            filters,
+		peerInfos:          cfg.PeerInfos,
+		subnetsProvider:    cfg.SubnetsProvider,
+		domainTypeProvider: cfg.DomainTypeProvider,
+		net:                cfg.Network,
 	}
 	return h
 }
@@ -103,7 +108,7 @@ func (h *handshaker) Handler(logger *zap.Logger) libp2pnetwork.StreamHandler {
 		}
 
 		// Respond with our own NodeInfo.
-		self, err := h.nodeInfos.SelfSealed()
+		self, err := h.sealedNodeRecord()
 		if err != nil {
 			return errors.Wrap(err, "could not seal self node info")
 		}
@@ -204,7 +209,7 @@ func (h *handshaker) updateNodeSubnets(logger *zap.Logger, pid peer.ID, ni *reco
 }
 
 func (h *handshaker) requestNodeInfo(logger *zap.Logger, conn libp2pnetwork.Conn) (*records.NodeInfo, error) {
-	data, err := h.nodeInfos.SelfSealed()
+	data, err := h.sealedNodeRecord()
 
 	if err != nil {
 		return nil, err
@@ -234,3 +239,14 @@ func (h *handshaker) applyFilters(sender peer.ID, ni *records.NodeInfo) error {
 
 	return nil
 }
+
+func (h *handshaker) sealedNodeRecord() ([]byte, error) {
+	// Update DomainType.
+	h.nodeInfos.UpdateSelfRecord(func(self *records.NodeInfo) *records.NodeInfo {
+		dt := h.domainTypeProvider.DomainType()
+		self.NetworkID = "0x" + hex.EncodeToString(dt[:])
+		return self
+	})
+
+	return h.nodeInfos.SelfSealed()
+}

network/peers/connections/mock/mock_node_info_idx.go

@@ -28,7 +28,7 @@ func (m NodeInfoIndex) Self() *records.NodeInfo {
 	panic("implement me")
 }
 
-func (m NodeInfoIndex) UpdateSelfRecord(newInfo *records.NodeInfo) {
+func (m NodeInfoIndex) UpdateSelfRecord(update func(self *records.NodeInfo) *records.NodeInfo) {
 	//TODO implement me
 	panic("implement me")
 }

network/peers/index.go

@@ -62,7 +62,7 @@ type NodeInfoIndex interface {
 	Self() *records.NodeInfo
 
 	// UpdateSelfRecord updating current self with new one
-	UpdateSelfRecord(newInfo *records.NodeInfo)
+	UpdateSelfRecord(update func(self *records.NodeInfo) *records.NodeInfo)
 
 	// SetNodeInfo updates the given peer with the NodeInfo.
 	SetNodeInfo(id peer.ID, node *records.NodeInfo)

network/peers/peers_index.go

@@ -94,28 +94,27 @@ func (pi *peersIndex) AtLimit(dir libp2pnetwork.Direction) bool {
 	return len(peers) > maxPeers
 }
 
-func (pi *peersIndex) UpdateSelfRecord(newSelf *records.NodeInfo) {
+func (pi *peersIndex) UpdateSelfRecord(update func(self *records.NodeInfo) *records.NodeInfo) {
 	pi.selfLock.Lock()
 	defer pi.selfLock.Unlock()
 
-	pi.self = newSelf
+	pi.self = update(pi.self.Clone())
 }
 
 func (pi *peersIndex) Self() *records.NodeInfo {
-	return pi.self
+	pi.selfLock.RLock()
+	defer pi.selfLock.RUnlock()
+
+	return pi.self.Clone()
 }
 
 func (pi *peersIndex) SelfSealed() ([]byte, error) {
-	pi.selfLock.Lock()
-	defer pi.selfLock.Unlock()
-
-	sealed, err := pi.self.Seal(pi.netKeyProvider())
+	sealed, err := pi.Self().Seal(pi.netKeyProvider())
 	if err != nil {
 		return nil, err
 	}
 
 	return sealed, nil
-
 }
 
 func (pi *peersIndex) SetNodeInfo(id peer.ID, nodeInfo *records.NodeInfo) {
@@ -148,7 +147,7 @@ func (pi *peersIndex) GetScore(id peer.ID, names ...string) ([]NodeScore, error)
 }
 
 func (pi *peersIndex) GetSubnetsStats() *SubnetsStats {
-	mySubnets, err := records.Subnets{}.FromString(pi.self.Metadata.Subnets)
+	mySubnets, err := records.Subnets{}.FromString(pi.Self().Metadata.Subnets)
 	if err != nil {
 		mySubnets, _ = records.Subnets{}.FromString(records.ZeroSubnets)
 	}

network/records/metadata.go

@@ -17,25 +17,19 @@ type NodeMetadata struct {
 }
 
 // Encode encodes the metadata into bytes
-// TODO: switch to SSZ
 func (nm *NodeMetadata) Encode() ([]byte, error) {
-	// ser := newSerializable(
-	//	nm.NodeVersion,
-	//	nm.ConsensusNode,
-	//	nm.ExecutionNode,
-	//)
-
 	return json.Marshal(nm)
 }
 
 // Decode decodes a raw payload into metadata
-// TODO: switch to SSZ
 func (nm *NodeMetadata) Decode(data []byte) error {
-	// var ser serializable
-
 	if err := json.Unmarshal(data, nm); err != nil {
 		return err
 	}
-
 	return nil
 }
+
+func (nm *NodeMetadata) Clone() *NodeMetadata {
+	cpy := *nm
+	return &cpy
+}

network/records/node_info.go

@@ -124,3 +124,10 @@ func (ni *NodeInfo) UnmarshalRecord(data []byte) error {
 
 	return nil
 }
+
+func (ni *NodeInfo) Clone() *NodeInfo {
+	return &NodeInfo{
+		NetworkID: ni.NetworkID,
+		Metadata:  ni.Metadata.Clone(),
+	}
+}

networkconfig/config.go

@@ -7,8 +7,8 @@ import (
 	"time"
 
 	"github.com/attestantio/go-eth2-client/spec/phase0"
-	spectypes "github.com/ssvlabs/ssv-spec/types"
 
+	spectypes "github.com/ssvlabs/ssv-spec/types"
 	"github.com/ssvlabs/ssv/protocol/v2/blockchain/beacon"
 )
 

networkconfig/holesky-stage.go

@@ -12,7 +12,7 @@ var HoleskyStage = NetworkConfig{
 	Name:                 "holesky-stage",
 	Beacon:               beacon.NewNetwork(spectypes.HoleskyNetwork),
 	GenesisDomainType:    [4]byte{0x00, 0x00, 0x31, 0x12},
-	AlanDomainType:       [4]byte{0x00, 0x00, 0x31, 0x99},
+	AlanDomainType:       [4]byte{0x00, 0x00, 0x31, 0x13},
 	GenesisEpoch:         1,
 	RegistrySyncOffset:   new(big.Int).SetInt64(84599),
 	RegistryContractAddr: "0x0d33801785340072C452b994496B19f196b7eE15",

operator/validator/controller.go

@@ -1180,46 +1180,10 @@ func hasNewValidators(before []phase0.ValidatorIndex, after []phase0.ValidatorIn
 	return false
 }
 
-// TODO alan: use spec when they fix bugs
-func TempBeaconVoteValueCheckF(
-	signer spectypes.BeaconSigner,
-	slot phase0.Slot,
-	sharePublicKey []byte,
-	estimatedCurrentEpoch phase0.Epoch,
-) specqbft.ProposedValueCheckF {
-	return func(data []byte) error {
-		bv := spectypes.BeaconVote{}
-		if err := bv.Decode(data); err != nil {
-			return errors.Wrap(err, "failed decoding beacon vote")
-		}
-
-		if bv.Target.Epoch > estimatedCurrentEpoch+1 {
-			return errors.New("attestation data target epoch is into far future")
-		}
-
-		if bv.Source.Epoch >= bv.Target.Epoch {
-			return errors.New("attestation data source >= target")
-		}
-
-		// attestationData := &phase0.AttestationData{
-		// 	Slot: slot,
-		// 	// CommitteeIndex doesn't matter for slashing checks
-		// 	Index:           0,
-		// 	BeaconBlockRoot: bv.BlockRoot,
-		// 	Source:          bv.Source,
-		// 	Target:          bv.Target,
-		// }
-
-		// TODO: (Alan) REVERT SLASHING CHECK
-		// return signer.IsAttestationSlashable(sharePublicKey, attestationData)
-		return nil
-	}
-}
-
 func SetupCommitteeRunners(
 	ctx context.Context,
 	options validator.Options,
-) func(slot phase0.Slot, shares map[phase0.ValidatorIndex]*spectypes.Share) *runner.CommitteeRunner {
+) validator.CommitteeRunnerFunc {
 	buildController := func(role spectypes.RunnerRole, valueCheckF specqbft.ProposedValueCheckF) *qbftcontroller.Controller {
 		config := &qbft.Config{
 			BeaconSigner:      options.Signer,
@@ -1245,10 +1209,10 @@ func SetupCommitteeRunners(
 		return qbftCtrl
 	}
 
-	return func(slot phase0.Slot, shares map[phase0.ValidatorIndex]*spectypes.Share) *runner.CommitteeRunner {
+	return func(slot phase0.Slot, shares map[phase0.ValidatorIndex]*spectypes.Share, slashableValidators []spectypes.ShareValidatorPK) *runner.CommitteeRunner {
 		// Create a committee runner.
 		epoch := options.NetworkConfig.Beacon.GetBeaconNetwork().EstimatedEpochAtSlot(slot)
-		valCheck := TempBeaconVoteValueCheckF(options.Signer, slot, options.SSVShare.Share.SharePubKey, epoch) // TODO: (Alan) fix slashing check (committee is not 1 pubkey)
+		valCheck := specssv.BeaconVoteValueCheckF(options.Signer, slot, slashableValidators, epoch)
 		crunner := runner.NewCommitteeRunner(
 			options.NetworkConfig,
 			shares,