Implement validator custody (#14948)

* Node info: Rename cache and mutex.

* Add `VALIDATOR_CUSTODY_REQUIREMENT` and `BALANCE_PER_ADDITIONAL_CUSTODY_GROUP`.

* Implement `ValidatorsCustodyRequirement`.

* Sync service: Add tracked validators cache.

* `dataColumnSidecarByRootRPCHandler`: Remove custody columns in logs.

* `dataColumnSidecarByRangeRPCHandler`: Remove custody columns in logs.

* `blobsFromStoredDataColumns`: Simplify.

Do not make any more a difference between "can theoretically reconstruct" and "can actually reconstruct".

* Implement validator custody.

* Fix Nishant's comment.

* Fix Nishant's commit.

Author: nalepae

beacon-chain/blockchain/options.go

@@ -4,6 +4,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/async/event"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
 	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db/filesystem"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/execution"
@@ -213,3 +214,10 @@ func WithSyncChecker(checker Checker) Option {
 		return nil
 	}
 }
+
+func WithCustodyInfo(custodyInfo *peerdas.CustodyInfo) Option {
+	return func(s *Service) error {
+		s.cfg.CustodyInfo = custodyInfo
+		return nil
+	}
+}

beacon-chain/blockchain/process_block.go

@@ -682,8 +682,12 @@ func (s *Service) areDataColumnsAvailable(ctx context.Context, root [32]byte, si
 	// https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.10/specs/fulu/das-core.md#custody-sampling
 	nodeID := s.cfg.P2P.NodeID()
 
+	// Prevent custody group count to change during the rest of the function.
+	s.cfg.CustodyInfo.Mut.RLock()
+	defer s.cfg.CustodyInfo.Mut.RUnlock()
+
 	// Get the custody group sampling size for the node.
-	custodyGroupSamplingSize := peerdas.CustodyGroupSamplingSize()
+	custodyGroupSamplingSize := s.cfg.CustodyInfo.CustodyGroupSamplingSize(peerdas.Actual)
 	peerInfo, _, err := peerdas.Info(nodeID, custodyGroupSamplingSize)
 	if err != nil {
 		return errors.Wrap(err, "peer info")

beacon-chain/blockchain/service.go

@@ -17,6 +17,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed"
 	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	coreTime "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/time"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/transition"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db"
@@ -93,6 +94,7 @@ type config struct {
 	FinalizedStateAtStartUp state.BeaconState
 	ExecutionEngineCaller   execution.EngineCaller
 	SyncChecker             Checker
+	CustodyInfo             *peerdas.CustodyInfo
 }
 
 // Checker is an interface used to determine if a node is in initial sync

beacon-chain/core/peerdas/BUILD.bazel

@@ -14,11 +14,13 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//beacon-chain/blockchain/kzg:go_default_library",
+        "//beacon-chain/state:go_default_library",
         "//cmd/beacon-chain/flags:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
         "//consensus-types/interfaces:go_default_library",
+        "//consensus-types/primitives:go_default_library",
         "//crypto/hash:go_default_library",
         "//encoding/bytesutil:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
@@ -46,10 +48,12 @@ go_test(
     deps = [
         ":go_default_library",
         "//beacon-chain/blockchain/kzg:go_default_library",
+        "//beacon-chain/state/state-native:go_default_library",
         "//cmd/beacon-chain/flags:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
+        "//consensus-types/primitives:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "//testing/require:go_default_library",
         "//testing/util:go_default_library",

beacon-chain/core/peerdas/das_core.go

@@ -11,10 +11,12 @@ import (
 	"github.com/holiman/uint256"
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/kzg"
+	beaconState "github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v5/crypto/hash"
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
@@ -30,6 +32,13 @@ var (
 	maxUint256 = &uint256.Int{math.MaxUint64, math.MaxUint64, math.MaxUint64, math.MaxUint64}
 )
 
+type CustodyType int
+
+const (
+	Target CustodyType = iota
+	Actual
+)
+
 // CustodyGroups computes the custody groups the node should participate in for custody.
 // https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.10/specs/fulu/das-core.md#get_custody_groups
 func CustodyGroups(nodeId enode.ID, custodyGroupCount uint64) (map[uint64]bool, error) {
@@ -193,10 +202,14 @@ func DataColumnSidecars(signedBlock interfaces.ReadOnlySignedBeaconBlock, blobs
 
 // CustodyGroupSamplingSize returns the number of custody groups the node should sample from.
 // https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.10/specs/fulu/das-core.md#custody-sampling
-func CustodyGroupSamplingSize() uint64 {
-	samplesPerSlot := params.BeaconConfig().SamplesPerSlot
-	custodyGroupCount := CustodyGroupCount()
+func (custodyInfo *CustodyInfo) CustodyGroupSamplingSize(ct CustodyType) uint64 {
+	custodyGroupCount := custodyInfo.TargetGroupCount.Get()
 
+	if ct == Actual {
+		custodyGroupCount = custodyInfo.ActualGroupCount()
+	}
+
+	samplesPerSlot := params.BeaconConfig().SamplesPerSlot
 	return max(samplesPerSlot, custodyGroupCount)
 }
 
@@ -226,6 +239,28 @@ func CustodyColumns(custodyGroups map[uint64]bool) (map[uint64]bool, error) {
 	return columns, nil
 }
 
+// ValidatorsCustodyRequirement returns the number of custody groups regarding the validator indices attached to the beacon node.
+// https://github.com/ethereum/consensus-specs/blob/dev/specs/fulu/das-core.md#validator-custody
+func ValidatorsCustodyRequirement(state beaconState.ReadOnlyBeaconState, validatorsIndex map[primitives.ValidatorIndex]bool) (uint64, error) {
+	totalNodeBalance := uint64(0)
+	for index := range validatorsIndex {
+		balance, err := state.BalanceAtIndex(index)
+		if err != nil {
+			return 0, errors.Wrapf(err, "balance at index for validator index %v", index)
+		}
+
+		totalNodeBalance += balance
+	}
+
+	beaconConfig := params.BeaconConfig()
+	numberOfCustodyGroup := beaconConfig.NumberOfCustodyGroups
+	validatorCustodyRequirement := beaconConfig.ValidatorCustodyRequirement
+	balancePerAdditionalCustodyGroup := beaconConfig.BalancePerAdditionalCustodyGroup
+
+	count := totalNodeBalance / balancePerAdditionalCustodyGroup
+	return min(max(count, validatorCustodyRequirement), numberOfCustodyGroup), nil
+}
+
 // Blobs extract blobs from `dataColumnsSidecar`.
 // This can be seen as the reciprocal function of DataColumnSidecars.
 // `dataColumnsSidecar` needs to contain the datacolumns corresponding to the non-extended matrix,

beacon-chain/core/peerdas/das_core_test.go

@@ -6,8 +6,10 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/kzg"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
+	state_native "github.com/prysmaticlabs/prysm/v5/beacon-chain/state/state-native"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 	"github.com/prysmaticlabs/prysm/v5/testing/util"
@@ -147,3 +149,99 @@ func TestDataColumnsSidecarsBlobsRoundtrip(t *testing.T) {
 	// Check that the blobs are the same.
 	require.DeepSSZEqual(t, verifiedROBlobs, roundtripBlobs)
 }
+
+func TestValidatorsCustodyRequirement(t *testing.T) {
+	testCases := []struct {
+		name     string
+		count    uint64
+		expected uint64
+	}{
+		{name: "0 validators", count: 0, expected: 8},
+		{name: "1 validator", count: 1, expected: 8},
+		{name: "8 validators", count: 8, expected: 8},
+		{name: "9 validators", count: 9, expected: 9},
+		{name: "100 validators", count: 100, expected: 100},
+		{name: "128 validators", count: 128, expected: 128},
+		{name: "129 validators", count: 129, expected: 128},
+		{name: "1000 validators", count: 1000, expected: 128},
+	}
+
+	const balance = uint64(32_000_000_000)
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			balances := make([]uint64, 0, tc.count)
+			for range tc.count {
+				balances = append(balances, balance)
+			}
+
+			validatorsIndex := make(map[primitives.ValidatorIndex]bool)
+			for i := range tc.count {
+				validatorsIndex[primitives.ValidatorIndex(i)] = true
+			}
+
+			beaconState, err := state_native.InitializeFromProtoFulu(&ethpb.BeaconStateElectra{Balances: balances})
+			require.NoError(t, err)
+
+			actual, err := peerdas.ValidatorsCustodyRequirement(beaconState, validatorsIndex)
+			require.NoError(t, err)
+			require.Equal(t, tc.expected, actual)
+		})
+	}
+}
+
+func TestCustodyGroupSamplingSize(t *testing.T) {
+	testCases := []struct {
+		name                         string
+		custodyType                  peerdas.CustodyType
+		validatorsCustodyRequirement uint64
+		toAdvertiseCustodyGroupCount uint64
+		expected                     uint64
+	}{
+		{
+			name:                         "target, lower than samples per slot",
+			custodyType:                  peerdas.Target,
+			validatorsCustodyRequirement: 2,
+			expected:                     8,
+		},
+		{
+			name:                         "target, higher than samples per slot",
+			custodyType:                  peerdas.Target,
+			validatorsCustodyRequirement: 100,
+			expected:                     100,
+		},
+		{
+			name:                         "actual, lower than samples per slot",
+			custodyType:                  peerdas.Actual,
+			validatorsCustodyRequirement: 3,
+			toAdvertiseCustodyGroupCount: 4,
+			expected:                     8,
+		},
+		{
+			name:                         "actual, higher than samples per slot",
+			custodyType:                  peerdas.Actual,
+			validatorsCustodyRequirement: 100,
+			toAdvertiseCustodyGroupCount: 101,
+			expected:                     100,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Create a custody info.
+			custodyInfo := peerdas.CustodyInfo{}
+
+			// Set the validators custody requirement for target custody group count.
+			custodyInfo.TargetGroupCount.SetValidatorsCustodyRequirement(tc.validatorsCustodyRequirement)
+
+			// Set the to advertise custody group count.
+			custodyInfo.ToAdvertiseGroupCount.Set(tc.toAdvertiseCustodyGroupCount)
+
+			// Compute the custody group sampling size.
+			actual := custodyInfo.CustodyGroupSamplingSize(tc.custodyType)
+
+			// Check the result.
+			require.Equal(t, tc.expected, actual)
+		})
+	}
+}