Avoid computing peerDAS info again and again. (#14893)

* `areDataColumnsAvailable`: `signed` ==> `signedBlock`.

* peerdas: Split `helpers.go` in multiple files respecting the specification.

* peerDAS: Implement `Info`.

* peerDAS: Use cached `Info` when possible.

beacon-chain/blockchain/process_block.go

@@ -645,12 +645,12 @@ func uint64MapToSortedSlice(input map[uint64]bool) []uint64 {
 	return output
 }
 
-func (s *Service) areDataColumnsAvailable(ctx context.Context, root [32]byte, signed interfaces.ReadOnlySignedBeaconBlock) error {
-	if signed.Version() < version.Fulu {
+func (s *Service) areDataColumnsAvailable(ctx context.Context, root [32]byte, signedBlock interfaces.ReadOnlySignedBeaconBlock) error {
+	if signedBlock.Version() < version.Fulu {
 		return nil
 	}
 
-	block := signed.Block()
+	block := signedBlock.Block()
 	if block == nil {
 		return errors.New("invalid nil beacon block")
 	}
@@ -681,24 +681,19 @@ func (s *Service) areDataColumnsAvailable(ctx context.Context, root [32]byte, si
 	// All columns to sample need to be available for the block to be considered available.
 	// https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.10/specs/fulu/das-core.md#custody-sampling
 	nodeID := s.cfg.P2P.NodeID()
-	custodyGroupSamplingSize := peerdas.CustodyGroupSamplingSize()
 
-	custodyGroups, err := peerdas.CustodyGroups(nodeID, custodyGroupSamplingSize)
+	// Get the custody group sampling size for the node.
+	custodyGroupSamplingSize := peerdas.CustodyGroupSamplingSize()
+	peerInfo, _, err := peerdas.Info(nodeID, custodyGroupSamplingSize)
 	if err != nil {
-		return errors.Wrap(err, "custody groups")
+		return errors.Wrap(err, "peer info")
 	}
 
 	// Exit early if the node is not expected to custody any data columns.
-	if len(custodyGroups) == 0 {
+	if len(peerInfo.CustodyColumns) == 0 {
 		return nil
 	}
 
-	// Get the custody columns from the groups.
-	columnsMap, err := peerdas.CustodyColumns(custodyGroups)
-	if err != nil {
-		return errors.Wrap(err, "custody columns")
-	}
-
 	// Subscribe to newsly data columns stored in the database.
 	rootIndexChan := make(chan filesystem.RootIndexPair)
 	subscription := s.blobStorage.DataColumnFeed.Subscribe(rootIndexChan)
@@ -722,7 +717,7 @@ func (s *Service) areDataColumnsAvailable(ctx context.Context, root [32]byte, si
 	}
 
 	// Get a map of data column indices that are not currently available.
-	missingMap, err := missingDataColumns(s.blobStorage, root, columnsMap)
+	missingMap, err := missingDataColumns(s.blobStorage, root, peerInfo.CustodyColumns)
 	if err != nil {
 		return err
 	}
@@ -734,7 +729,7 @@ func (s *Service) areDataColumnsAvailable(ctx context.Context, root [32]byte, si
 	}
 
 	// Log for DA checks that cross over into the next slot; helpful for debugging.
-	nextSlot := slots.BeginsAt(signed.Block().Slot()+1, s.genesisTime)
+	nextSlot := slots.BeginsAt(signedBlock.Block().Slot()+1, s.genesisTime)
 	// Avoid logging if DA check is called after next slot start.
 	if nextSlot.After(time.Now()) {
 		nst := time.AfterFunc(time.Until(nextSlot), func() {
@@ -750,18 +745,18 @@ func (s *Service) areDataColumnsAvailable(ctx context.Context, root [32]byte, si
 			)
 
 			numberOfColumns := params.BeaconConfig().NumberOfColumns
-			colMapCount := uint64(len(columnsMap))
+			colMapCount := uint64(len(peerInfo.CustodyColumns))
 
 			if colMapCount < numberOfColumns {
-				expected = uint64MapToSortedSlice(columnsMap)
+				expected = uint64MapToSortedSlice(peerInfo.CustodyColumns)
 			}
 
 			if missingMapCount < numberOfColumns {
 				missing = uint64MapToSortedSlice(missingMap)
 			}
 
 			log.WithFields(logrus.Fields{
-				"slot":            signed.Block().Slot(),
+				"slot":            signedBlock.Block().Slot(),
 				"root":            fmt.Sprintf("%#x", root),
 				"columnsExpected": expected,
 				"columnsWaiting":  missing,

beacon-chain/cache/BUILD.bazel

@@ -9,7 +9,6 @@ go_library(
         "attestation_data.go",
         "balance_cache_key.go",
         "checkpoint_state.go",
-        "column_subnet_ids.go",
         "committee.go",
         "committee_disabled.go",  # keep
         "committees.go",

beacon-chain/core/peerdas/BUILD.bazel

@@ -3,9 +3,12 @@ load("@prysm//tools/go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
-        "helpers.go",
-        "log.go",
+        "das_core.go",
+        "info.go",
         "metrics.go",
+        "p2p_interface.go",
+        "peer_sampling.go",
+        "reconstruction.go",
     ],
     importpath = "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas",
     visibility = ["//visibility:public"],
@@ -21,18 +24,25 @@ go_library(
         "//proto/prysm/v1alpha1:go_default_library",
         "@com_github_ethereum_go_ethereum//p2p/enode:go_default_library",
         "@com_github_ethereum_go_ethereum//p2p/enr:go_default_library",
+        "@com_github_hashicorp_golang_lru//:go_default_library",
         "@com_github_holiman_uint256//:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
         "@org_golang_x_sync//errgroup:go_default_library",
     ],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = ["helpers_test.go"],
+    srcs = [
+        "das_core_test.go",
+        "info_test.go",
+        "p2p_interface_test.go",
+        "peer_sampling_test.go",
+        "reconstruction_test.go",
+        "utils_test.go",
+    ],
     deps = [
         ":go_default_library",
         "//beacon-chain/blockchain/kzg:go_default_library",
@@ -45,7 +55,9 @@ go_test(
         "//testing/util:go_default_library",
         "@com_github_consensys_gnark_crypto//ecc/bls12-381/fr:go_default_library",
         "@com_github_crate_crypto_go_kzg_4844//:go_default_library",
+        "@com_github_ethereum_go_ethereum//p2p/enode:go_default_library",
         "@com_github_ethereum_go_ethereum//p2p/enr:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
     ],
 )