PeerDAS: Multiple improvements (#14467)

* `scheduleReconstructedDataColumnsBroadcast`: Really minor refactor.

* `receivedDataColumnsFromRootLock` -> `dataColumnsFromRootLock`

* `reconstructDataColumns`: Stop looking into the DB to know if we have some columns.

Before this commit:
Each time we receive a column, we look into the filesystem for all columns we store.
==> For 128 columns, it looks for 1 + 2 + 3 + ... + 128 = 128(128+1)/2 = 8256 files look.

Also, as soon as a column is saved into the file system, then if, right after, we
look at the filesystem again, we assume the column will be available (strict consistency).
It happens not to be always true.

==> Sometimes, we can reconstruct and reseed columns more than once, because of this lack of filesystem strict consistency.

After this commit:
We use a (strictly consistent) cache to determine if we received a column or not.
==> No more consistency issue, and less stress for the filesystem.

* `dataColumnSidecarByRootRPCHandler`: Improve logging.

Before this commit, logged values assumed that all requested columns correspond to
the same block root, which is not always the case.

After this commit, we know which columns are requested for which root.

* Add a log when broadcasting a data column.

This is useful to debug "lost data columns" in devnet.

* Address Nishant's comment

beacon-chain/blockchain/setup_test.go

@@ -22,6 +22,7 @@ import (
 	p2pTesting "github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/testing"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/startup"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state/stategen"
+	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 	"google.golang.org/protobuf/proto"
@@ -71,7 +72,7 @@ func (mb *mockBroadcaster) BroadcastBlob(_ context.Context, _ uint64, _ *ethpb.B
 	return nil
 }
 
-func (mb *mockBroadcaster) BroadcastDataColumn(_ context.Context, _ uint64, _ *ethpb.DataColumnSidecar) error {
+func (mb *mockBroadcaster) BroadcastDataColumn(_ context.Context, _ [fieldparams.RootLength]byte, _ uint64, _ *ethpb.DataColumnSidecar) error {
 	mb.broadcastCalled = true
 	return nil
 }

beacon-chain/p2p/broadcaster.go

@@ -282,7 +282,12 @@ func (s *Service) internalBroadcastBlob(
 // BroadcastDataColumn broadcasts a data column to the p2p network, the message is assumed to be
 // broadcasted to the current fork and to the input column subnet.
 // TODO: Add tests
-func (s *Service) BroadcastDataColumn(ctx context.Context, columnSubnet uint64, dataColumnSidecar *ethpb.DataColumnSidecar) error {
+func (s *Service) BroadcastDataColumn(
+	ctx context.Context,
+	root [fieldparams.RootLength]byte,
+	columnSubnet uint64,
+	dataColumnSidecar *ethpb.DataColumnSidecar,
+) error {
 	// Add tracing to the function.
 	ctx, span := trace.StartSpan(ctx, "p2p.BroadcastBlob")
 	defer span.End()
@@ -301,13 +306,14 @@ func (s *Service) BroadcastDataColumn(ctx context.Context, columnSubnet uint64,
 	}
 
 	// Non-blocking broadcast, with attempts to discover a column subnet peer if none available.
-	go s.internalBroadcastDataColumn(ctx, columnSubnet, dataColumnSidecar, forkDigest)
+	go s.internalBroadcastDataColumn(ctx, root, columnSubnet, dataColumnSidecar, forkDigest)
 
 	return nil
 }
 
 func (s *Service) internalBroadcastDataColumn(
 	ctx context.Context,
+	root [fieldparams.RootLength]byte,
 	columnSubnet uint64,
 	dataColumnSidecar *ethpb.DataColumnSidecar,
 	forkDigest [fieldparams.VersionLength]byte,
@@ -368,6 +374,21 @@ func (s *Service) internalBroadcastDataColumn(
 		tracing.AnnotateError(span, err)
 	}
 
+	header := dataColumnSidecar.SignedBlockHeader.GetHeader()
+	slot := header.GetSlot()
+
+	slotStartTime, err := slots.ToTime(uint64(s.genesisTime.Unix()), slot)
+	if err != nil {
+		log.WithError(err).Error("Failed to convert slot to time")
+	}
+
+	log.WithFields(logrus.Fields{
+		"slot":               slot,
+		"timeSinceSlotStart": time.Since(slotStartTime),
+		"root":               fmt.Sprintf("%#x", root),
+		"columnSubnet":       columnSubnet,
+	}).Debug("Broadcasted data column sidecar")
+
 	// Increase the number of successful broadcasts.
 	dataColumnSidecarBroadcasts.Inc()
 }

beacon-chain/p2p/broadcaster_test.go

@@ -585,9 +585,10 @@ func TestService_BroadcastDataColumn(t *testing.T) {
 
 	// Attempt to broadcast nil object should fail.
 	ctx := context.Background()
-	require.ErrorContains(t, "attempted to broadcast nil", p.BroadcastDataColumn(ctx, subnet, nil))
+	var root [fieldparams.RootLength]byte
+	require.ErrorContains(t, "attempted to broadcast nil", p.BroadcastDataColumn(ctx, root, subnet, nil))
 
 	// Broadcast to peers and wait.
-	require.NoError(t, p.BroadcastDataColumn(ctx, subnet, sidecar))
+	require.NoError(t, p.BroadcastDataColumn(ctx, root, subnet, sidecar))
 	require.Equal(t, false, util.WaitTimeout(&wg, 1*time.Second), "Failed to receive pubsub within 1s")
 }

beacon-chain/p2p/interfaces.go

@@ -13,6 +13,7 @@ import (
 	"github.com/multiformats/go-multiaddr"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/encoder"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/peers"
+	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1/metadata"
 	"google.golang.org/protobuf/proto"
@@ -43,7 +44,7 @@ type Broadcaster interface {
 	BroadcastAttestation(ctx context.Context, subnet uint64, att ethpb.Att) error
 	BroadcastSyncCommitteeMessage(ctx context.Context, subnet uint64, sMsg *ethpb.SyncCommitteeMessage) error
 	BroadcastBlob(ctx context.Context, subnet uint64, blob *ethpb.BlobSidecar) error
-	BroadcastDataColumn(ctx context.Context, columnSubnet uint64, dataColumnSidecar *ethpb.DataColumnSidecar) error
+	BroadcastDataColumn(ctx context.Context, root [fieldparams.RootLength]byte, columnSubnet uint64, dataColumnSidecar *ethpb.DataColumnSidecar) error
 }
 
 // SetStreamHandler configures p2p to handle streams of a certain topic ID.

beacon-chain/p2p/testing/BUILD.bazel

@@ -21,6 +21,7 @@ go_library(
         "//beacon-chain/p2p/encoder:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/peers/scorers:go_default_library",
+        "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "//proto/prysm/v1alpha1/metadata:go_default_library",

beacon-chain/p2p/testing/fuzz_p2p.go

@@ -13,6 +13,7 @@ import (
 	"github.com/multiformats/go-multiaddr"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/encoder"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/peers"
+	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1/metadata"
 	"google.golang.org/protobuf/proto"
@@ -155,7 +156,7 @@ func (*FakeP2P) BroadcastBlob(_ context.Context, _ uint64, _ *ethpb.BlobSidecar)
 }
 
 // BroadcastDataColumn -- fake.
-func (*FakeP2P) BroadcastDataColumn(_ context.Context, _ uint64, _ *ethpb.DataColumnSidecar) error {
+func (*FakeP2P) BroadcastDataColumn(_ context.Context, _ [fieldparams.RootLength]byte, _ uint64, _ *ethpb.DataColumnSidecar) error {
 	return nil
 }
 

beacon-chain/p2p/testing/mock_broadcaster.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"google.golang.org/protobuf/proto"
 )
@@ -49,7 +50,7 @@ func (m *MockBroadcaster) BroadcastBlob(context.Context, uint64, *ethpb.BlobSide
 }
 
 // BroadcastDataColumn broadcasts a data column for mock.
-func (m *MockBroadcaster) BroadcastDataColumn(context.Context, uint64, *ethpb.DataColumnSidecar) error {
+func (m *MockBroadcaster) BroadcastDataColumn(context.Context, [fieldparams.RootLength]byte, uint64, *ethpb.DataColumnSidecar) error {
 	m.BroadcastCalled.Store(true)
 	return nil
 }

beacon-chain/p2p/testing/p2p.go

@@ -28,6 +28,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/encoder"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/peers"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/peers/scorers"
+	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1/metadata"
@@ -207,7 +208,7 @@ func (p *TestP2P) BroadcastBlob(context.Context, uint64, *ethpb.BlobSidecar) err
 }
 
 // BroadcastDataColumn broadcasts a data column for mock.
-func (p *TestP2P) BroadcastDataColumn(context.Context, uint64, *ethpb.DataColumnSidecar) error {
+func (p *TestP2P) BroadcastDataColumn(context.Context, [fieldparams.RootLength]byte, uint64, *ethpb.DataColumnSidecar) error {
 	p.BroadcastCalled.Store(true)
 	return nil
 }

beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go

@@ -471,7 +471,6 @@ func (vs *Server) broadcastAndReceiveDataColumns(
 	slot primitives.Slot,
 ) error {
 	eg, _ := errgroup.WithContext(ctx)
-
 	dataColumnsWithholdCount := features.Get().DataColumnsWithholdCount
 
 	for _, sd := range sidecars {
@@ -487,11 +486,10 @@ func (vs *Server) broadcastAndReceiveDataColumns(
 				log.WithFields(logrus.Fields{
 					"root":            fmt.Sprintf("%#x", root),
 					"slot":            slot,
-					"subnet":          subnet,
 					"dataColumnIndex": sidecar.ColumnIndex,
 				}).Warning("Withholding data column")
 			} else {
-				if err := vs.P2P.BroadcastDataColumn(ctx, subnet, sidecar); err != nil {
+				if err := vs.P2P.BroadcastDataColumn(ctx, root, subnet, sidecar); err != nil {
 					return errors.Wrap(err, "broadcast data column")
 				}
 			}

beacon-chain/sync/BUILD.bazel

@@ -162,6 +162,7 @@ go_test(
         "block_batcher_test.go",
         "broadcast_bls_changes_test.go",
         "context_test.go",
+        "data_columns_reconstruct_test.go",
         "data_columns_sampling_test.go",
         "decode_pubsub_test.go",
         "error_test.go",