Fix orphaned blocks (#15720)

* `Broadcasted data column sidecar` log: Add `blobCount`.

* `broadcastAndReceiveDataColumns`: Broadcast and receive data columns in parallel.

* `ProposeBeaconBlock`: First broadcast/receive block, and then sidecars.

* `broadcastReceiveBlock`: Add log.

* Add changelog

* Fix deadlock-option 1.

* Fix deadlock-option 2.

* Take notifier out of the critical section

* only compute common info once, for all sidecars

---------

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>

beacon-chain/core/peerdas/validator.go

@@ -109,15 +109,14 @@ func DataColumnSidecars(rows []kzg.CellsAndProofs, src ConstructionPopulator) ([
 	if err != nil {
 		return nil, errors.Wrap(err, "rotate cells and proofs")
 	}
+	info, err := src.extract()
+	if err != nil {
+		return nil, errors.Wrap(err, "extract block info")
+	}
 
 	maxIdx := params.BeaconConfig().NumberOfColumns
 	roSidecars := make([]blocks.RODataColumn, 0, maxIdx)
 	for idx := range maxIdx {
-		info, err := src.extract()
-		if err != nil {
-			return nil, errors.Wrap(err, "extract block info")
-		}
-
 		sidecar := &ethpb.DataColumnSidecar{
 			Index:                        idx,
 			Column:                       cells[idx],

beacon-chain/p2p/broadcaster.go

@@ -381,6 +381,7 @@ func (s *Service) internalBroadcastDataColumnSidecar(
 		"timeSinceSlotStart": time.Since(slotStartTime),
 		"root":               fmt.Sprintf("%#x", dataColumnSidecar.BlockRoot()),
 		"columnSubnet":       columnSubnet,
+		"blobCount":          len(dataColumnSidecar.Column),
 	}).Debug("Broadcasted data column sidecar")
 
 	// Increase the number of successful broadcasts.

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

@@ -324,18 +324,18 @@ func (vs *Server) ProposeBeaconBlock(ctx context.Context, req *ethpb.GenericSign
 
 	wg.Add(1)
 	go func() {
-		defer wg.Done()
-		if err := vs.broadcastReceiveBlock(ctx, block, root); err != nil {
+		if err := vs.broadcastReceiveBlock(ctx, &wg, block, root); err != nil {
 			errChan <- errors.Wrap(err, "broadcast/receive block failed")
 			return
 		}
 		errChan <- nil
 	}()
 
+	wg.Wait()
+
 	if err := vs.broadcastAndReceiveSidecars(ctx, block, root, blobSidecars, dataColumnSidecars); err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not broadcast/receive sidecars: %v", err)
 	}
-	wg.Wait()
 	if err := <-errChan; err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not broadcast/receive block: %v", err)
 	}
@@ -432,7 +432,26 @@ func (vs *Server) handleUnblindedBlock(
 }
 
 // broadcastReceiveBlock broadcasts a block and handles its reception.
-func (vs *Server) broadcastReceiveBlock(ctx context.Context, block interfaces.SignedBeaconBlock, root [fieldparams.RootLength]byte) error {
+func (vs *Server) broadcastReceiveBlock(ctx context.Context, wg *sync.WaitGroup, block interfaces.SignedBeaconBlock, root [fieldparams.RootLength]byte) error {
+	if err := vs.broadcastBlock(ctx, wg, block, root); err != nil {
+		return errors.Wrap(err, "broadcast block")
+	}
+
+	vs.BlockNotifier.BlockFeed().Send(&feed.Event{
+		Type: blockfeed.ReceivedBlock,
+		Data: &blockfeed.ReceivedBlockData{SignedBlock: block},
+	})
+
+	if err := vs.BlockReceiver.ReceiveBlock(ctx, block, root, nil); err != nil {
+		return errors.Wrap(err, "receive block")
+	}
+
+	return nil
+}
+
+func (vs *Server) broadcastBlock(ctx context.Context, wg *sync.WaitGroup, block interfaces.SignedBeaconBlock, root [fieldparams.RootLength]byte) error {
+	defer wg.Done()
+
 	protoBlock, err := block.Proto()
 	if err != nil {
 		return errors.Wrap(err, "protobuf conversion failed")
@@ -440,11 +459,13 @@ func (vs *Server) broadcastReceiveBlock(ctx context.Context, block interfaces.Si
 	if err := vs.P2P.Broadcast(ctx, protoBlock); err != nil {
 		return errors.Wrap(err, "broadcast failed")
 	}
-	vs.BlockNotifier.BlockFeed().Send(&feed.Event{
-		Type: blockfeed.ReceivedBlock,
-		Data: &blockfeed.ReceivedBlockData{SignedBlock: block},
-	})
-	return vs.BlockReceiver.ReceiveBlock(ctx, block, root, nil)
+
+	log.WithFields(logrus.Fields{
+		"slot": block.Block().Slot(),
+		"root": fmt.Sprintf("%#x", root),
+	}).Debug("Broadcasted block")
+
+	return nil
 }
 
 // broadcastAndReceiveBlobs handles the broadcasting and reception of blob sidecars.
@@ -498,10 +519,6 @@ func (vs *Server) broadcastAndReceiveDataColumns(
 		})
 	}
 
-	if err := eg.Wait(); err != nil {
-		return errors.Wrap(err, "wait for data columns to be broadcasted")
-	}
-
 	if err := vs.DataColumnReceiver.ReceiveDataColumns(verifiedRODataColumns); err != nil {
 		return errors.Wrap(err, "receive data column")
 	}
@@ -512,6 +529,11 @@ func (vs *Server) broadcastAndReceiveDataColumns(
 			Data: &operation.DataColumnSidecarReceivedData{DataColumn: &verifiedRODataColumn}, // #nosec G601
 		})
 	}
+
+	if err := eg.Wait(); err != nil {
+		return errors.Wrap(err, "wait for data columns to be broadcasted")
+	}
+
 	return nil
 }
 

changelog/manu-broadcast.md

@@ -0,0 +1,4 @@
+### Changed
+- Broadcast block then sidecars, instead block and sidecars concurrently
+- Broadcast and receive sidecars in concurrently, instead sequentially
+