RefreshPersistentSubnets: Manage peerDAS.

beacon-chain/core/helpers/sync_committee_test.go

@@ -78,6 +78,7 @@ func TestIsCurrentEpochSyncCommittee_UsingCommittee(t *testing.T) {
 
 func TestIsCurrentEpochSyncCommittee_DoesNotExist(t *testing.T) {
 	helpers.ClearCache()
+	params.SetupTestConfigCleanup(t)
 
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := &ethpb.SyncCommittee{
@@ -264,6 +265,7 @@ func TestCurrentEpochSyncSubcommitteeIndices_UsingCommittee(t *testing.T) {
 }
 
 func TestCurrentEpochSyncSubcommitteeIndices_DoesNotExist(t *testing.T) {
+	params.SetupTestConfigCleanup(t)
 	helpers.ClearCache()
 
 	validators := make([]*ethpb.Validator, params.BeaconConfig().SyncCommitteeSize)

beacon-chain/p2p/discovery.go

@@ -105,7 +105,7 @@ func (s *Service) RefreshPersistentSubnets() {
 	// Is our attestation bitvector record up to date?
 	isBitVUpToDate := bytes.Equal(bitV, inRecordBitV) && bytes.Equal(bitV, inMetadataBitV)
 
-	// Compare current epoch with our fork epochs
+	// Compare current epoch with Altair fork epoch
 	altairForkEpoch := params.BeaconConfig().AltairForkEpoch
 
 	if currentEpoch < altairForkEpoch {
@@ -120,6 +120,7 @@ func (s *Service) RefreshPersistentSubnets() {
 
 		// Ping all peers.
 		s.pingPeers()
+
 		return
 	}
 
@@ -131,7 +132,7 @@ func (s *Service) RefreshPersistentSubnets() {
 	}
 
 	// Get the sync subnet bitfield we store in our record.
-	currentBitS, err := syncBitvector(record)
+	inRecordBitS, err := syncBitvector(record)
 	if err != nil {
 		log.WithError(err).Error("Could not retrieve sync bitfield")
 		return
@@ -140,15 +141,49 @@ func (s *Service) RefreshPersistentSubnets() {
 	// Get the sync subnet bitfield in our metadata.
 	currentBitSInMetadata := s.Metadata().SyncnetsBitfield()
 
-	isBitSUpToDate := bytes.Equal(bitS, currentBitS) && bytes.Equal(bitS, currentBitSInMetadata)
+	isBitSUpToDate := bytes.Equal(bitS, inRecordBitS) && bytes.Equal(bitS, currentBitSInMetadata)
+
+	// Compare current epoch with EIP-7594 fork epoch.
+	eip7594ForkEpoch := params.BeaconConfig().Eip7594ForkEpoch
+
+	if currentEpoch < eip7594ForkEpoch {
+		// Altair behaviour.
+		if metadataVersion == version.Altair && isBitVUpToDate && isBitSUpToDate {
+			// Nothing to do, return early.
+			return
+		}
+
+		// Some data have changed, update our record and metadata.
+		s.updateSubnetRecordWithMetadataV2(bitV, bitS)
+
+		// Ping all peers to inform them of new metadata
+		s.pingPeers()
+
+		return
+	}
+
+	// Get the current custody subnet count.
+	custodySubnetCount := peerdas.CustodySubnetCount()
+
+	// Get the custody subnet count we store in our record.
+	inRecordCustodySubnetCount, err := peerdas.CustodyCountFromRecord(record)
+	if err != nil {
+		log.WithError(err).Error("Could not retrieve custody subnet count")
+		return
+	}
+
+	// Get the custody subnet count in our metadata.
+	inMetadataCustodySubnetCount := s.Metadata().CustodySubnetCount()
+
+	isCustodySubnetCountUpToDate := (custodySubnetCount == inRecordCustodySubnetCount && custodySubnetCount == inMetadataCustodySubnetCount)
 
-	if metadataVersion == version.Altair && isBitVUpToDate && isBitSUpToDate {
+	if metadataVersion == version.Deneb && isBitVUpToDate && isBitSUpToDate && isCustodySubnetCountUpToDate {
 		// Nothing to do, return early.
 		return
 	}
 
 	// Some data changed. Update the record and the metadata.
-	s.updateSubnetRecordWithMetadataV2(bitV, bitS)
+	s.updateSubnetRecordWithMetadataV3(bitV, bitS, custodySubnetCount)
 
 	// Ping all peers.
 	s.pingPeers()

beacon-chain/p2p/discovery_test.go

@@ -473,6 +473,7 @@ type check struct {
 	metadataSequenceNumber uint64
 	attestationSubnets     []uint64
 	syncSubnets            []uint64
+	custodySubnetCount     *uint64
 }
 
 func checkPingCountCacheMetadataRecord(
@@ -537,15 +538,41 @@ func checkPingCountCacheMetadataRecord(
 
 		actualBitSMetadata := service.metaData.SyncnetsBitfield()
 		require.DeepSSZEqual(t, expectedBitS, actualBitSMetadata)
+	}
+
+	if expected.custodySubnetCount != nil {
+		// Check custody subnet count in ENR.
+		var actualCustodySubnetCount uint64
+		err := service.dv5Listener.LocalNode().Node().Record().Load(enr.WithEntry(peerdas.CustodySubnetCountEnrKey, &actualCustodySubnetCount))
+		require.NoError(t, err)
+		require.Equal(t, *expected.custodySubnetCount, actualCustodySubnetCount)
 
+		// Check custody subnet count in metadata.
+		actualCustodySubnetCountMetadata := service.metaData.CustodySubnetCount()
+		require.Equal(t, *expected.custodySubnetCount, actualCustodySubnetCountMetadata)
 	}
 }
 
 func TestRefreshPersistentSubnets(t *testing.T) {
-	const altairForkEpoch = 5
+	params.SetupTestConfigCleanup(t)
+
+	// Clean up caches after usage.
+	defer cache.SubnetIDs.EmptyAllCaches()
+	defer cache.SyncSubnetIDs.EmptyAllCaches()
+
+	const (
+		altairForkEpoch  = 5
+		eip7594ForkEpoch = 10
+	)
+
+	custodySubnetCount := uint64(1)
 
 	// Set up epochs.
-	params.BeaconConfig().AltairForkEpoch = altairForkEpoch
+	defaultCfg := params.BeaconConfig()
+	cfg := defaultCfg.Copy()
+	cfg.AltairForkEpoch = altairForkEpoch
+	cfg.Eip7594ForkEpoch = eip7594ForkEpoch
+	params.OverrideBeaconConfig(cfg)
 
 	// Compute the number of seconds per epoch.
 	secondsPerSlot := params.BeaconConfig().SecondsPerSlot
@@ -613,6 +640,39 @@ func TestRefreshPersistentSubnets(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:              "PeerDAS",
+			epochSinceGenesis: eip7594ForkEpoch,
+			checks: []check{
+				{
+					pingCount:              0,
+					metadataSequenceNumber: 0,
+					attestationSubnets:     []uint64{},
+					syncSubnets:            nil,
+				},
+				{
+					pingCount:              1,
+					metadataSequenceNumber: 1,
+					attestationSubnets:     []uint64{40, 41},
+					syncSubnets:            nil,
+					custodySubnetCount:     &custodySubnetCount,
+				},
+				{
+					pingCount:              2,
+					metadataSequenceNumber: 2,
+					attestationSubnets:     []uint64{40, 41},
+					syncSubnets:            []uint64{1, 2},
+					custodySubnetCount:     &custodySubnetCount,
+				},
+				{
+					pingCount:              2,
+					metadataSequenceNumber: 2,
+					attestationSubnets:     []uint64{40, 41},
+					syncSubnets:            []uint64{1, 2},
+					custodySubnetCount:     &custodySubnetCount,
+				},
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -693,4 +753,7 @@ func TestRefreshPersistentSubnets(t *testing.T) {
 			cache.SyncSubnetIDs.EmptyAllCaches()
 		})
 	}
+
+	// Reset the config.
+	params.OverrideBeaconConfig(defaultCfg)
 }

beacon-chain/p2p/subnets.go

@@ -31,6 +31,7 @@ var syncCommsSubnetCount = params.BeaconConfig().SyncCommitteeSubnetCount
 
 var attSubnetEnrKey = params.BeaconNetworkConfig().AttSubnetKey
 var syncCommsSubnetEnrKey = params.BeaconNetworkConfig().SyncCommsSubnetKey
+var custodySubnetCountEnrKey = params.BeaconNetworkConfig().CustodySubnetCountKey
 
 // The value used with the subnet, in order
 // to create an appropriate key to retrieve
@@ -219,6 +220,35 @@ func (s *Service) updateSubnetRecordWithMetadataV2(bitVAtt bitfield.Bitvector64,
 	})
 }
 
+// updateSubnetRecordWithMetadataV3 updates:
+// - attestation subnet tracked,
+// - sync subnets tracked, and
+// - custody subnet count
+// both in the node's record and in the node's metadata.
+func (s *Service) updateSubnetRecordWithMetadataV3(
+	bitVAtt bitfield.Bitvector64,
+	bitVSync bitfield.Bitvector4,
+	custodySubnetCount uint64,
+) {
+	attSubnetsEntry := enr.WithEntry(attSubnetEnrKey, &bitVAtt)
+	syncSubnetsEntry := enr.WithEntry(syncCommsSubnetEnrKey, &bitVSync)
+	custodySubnetCountEntry := enr.WithEntry(custodySubnetCountEnrKey, custodySubnetCount)
+
+	localNode := s.dv5Listener.LocalNode()
+	localNode.Set(attSubnetsEntry)
+	localNode.Set(syncSubnetsEntry)
+	localNode.Set(custodySubnetCountEntry)
+
+	newSeqNumber := s.metaData.SequenceNumber() + 1
+
+	s.metaData = wrapper.WrappedMetadataV2(&pb.MetaDataV2{
+		SeqNumber:          newSeqNumber,
+		Attnets:            bitVAtt,
+		Syncnets:           bitVSync,
+		CustodySubnetCount: custodySubnetCount,
+	})
+}
+
 func initializePersistentSubnets(id enode.ID, epoch primitives.Epoch) error {
 	_, ok, expTime := cache.SubnetIDs.GetPersistentSubnets()
 	if ok && expTime.After(time.Now()) {

config/params/mainnet_config.go

@@ -35,6 +35,7 @@ var mainnetNetworkConfig = &NetworkConfig{
 	ETH2Key:                    "eth2",
 	AttSubnetKey:               "attnets",
 	SyncCommsSubnetKey:         "syncnets",
+	CustodySubnetCountKey:      "csc",
 	MinimumPeersInSubnetSearch: 20,
 	ContractDeploymentBlock:    11184524, // Note: contract was deployed in block 11052984 but no transactions were sent until 11184524.
 	BootstrapNodes: []string{

config/params/network_config.go

@@ -8,9 +8,10 @@ import (
 // NetworkConfig defines the spec based network parameters.
 type NetworkConfig struct {
 	// DiscoveryV5 Config
-	ETH2Key                    string // ETH2Key is the ENR key of the Ethereum consensus object in an enr.
-	AttSubnetKey               string // AttSubnetKey is the ENR key of the subnet bitfield in the enr.
-	SyncCommsSubnetKey         string // SyncCommsSubnetKey is the ENR key of the sync committee subnet bitfield in the enr.
+	ETH2Key                    string // ETH2Key is the ENR key of the Ethereum consensus object.
+	AttSubnetKey               string // AttSubnetKey is the ENR key of the subnet bitfield.
+	SyncCommsSubnetKey         string // SyncCommsSubnetKey is the ENR key of the sync committee subnet bitfield.
+	CustodySubnetCountKey      string // CustodySubnetCountKey is the ENR key of the custody subnet count.
 	MinimumPeersInSubnetSearch uint64 // PeersInSubnetSearch is the required amount of peers that we need to be able to lookup in a subnet search.
 
 	// Chain Network Config