Various small optimizations (#15153)

* Reconstruct data columns from gossip source: Call `setSeenDataColumnIndex`.

* `reconstructAndBroadcastDataColumnSidecars`: Minor optimisation.

Avoid to range over all columns.

* Reconstructed data columns sidecars from EL: Avoid broadcasting already received data columns.

Author: nalepae

beacon-chain/sync/data_columns_reconstruct.go

@@ -109,6 +109,8 @@ func (s *Service) reconstructDataColumns(ctx context.Context, verifiedRODataColu
 	}
 
 	// Save the data columns sidecars in the database.
+	// Note: We do not call `receiveDataColumn`, because it will ignore
+	// incoming data columns via gossip while we did not broadcast (yet) the reconstructed data columns.
 	if err := s.cfg.dataColumnStorage.Save(verifiedRODataColumns); err != nil {
 		return errors.Wrap(err, "save data column sidecars")
 	}
@@ -214,6 +216,9 @@ func (s *Service) scheduleReconstructedDataColumnsBroadcast(
 			if err := s.cfg.p2p.BroadcastDataColumn(ctx, root, subnet, verifiedRODataColumn.DataColumnSidecar); err != nil {
 				log.WithError(err).Error("Broadcast data column")
 			}
+
+			// Now, we can set the data column as seen.
+			s.setSeenDataColumnIndex(slot, proposerIndex, verifiedRODataColumn.Index)
 		}
 
 		// Sort the missing data columns.

beacon-chain/sync/subscriber_beacon_blocks.go

@@ -118,12 +118,23 @@ func (s *Service) reconstructAndBroadcastDataColumnSidecars(ctx context.Context,
 		return
 	}
 
-	// Broadcast data column and then save to db (if needs to be in custody)
-	for _, sidecar := range sidecars {
-		if !info.CustodyColumns[sidecar.Index] {
+	blockSlot := block.Slot()
+	proposerIndex := block.ProposerIndex()
+
+	// Broadcast and save data columns sidecars to custody but not yet received.
+	sidecarCount := uint64(len(sidecars))
+	for columnIndex := range info.CustodyColumns {
+		if columnIndex >= sidecarCount {
+			log.WithField("index", columnIndex).Error("Sidecar index out of range - should never happen")
 			continue
 		}
 
+		if s.hasSeenDataColumnIndex(blockSlot, proposerIndex, columnIndex) {
+			continue
+		}
+
+		sidecar := sidecars[columnIndex]
+
 		if err := s.cfg.p2p.BroadcastDataColumn(ctx, blockRoot, sidecar.Index, sidecar.DataColumnSidecar); err != nil {
 			log.WithFields(dataColumnFields(sidecar.RODataColumn)).WithError(err).Error("Failed to broadcast data column")
 		}