PeerDAS: Withhold data on purpose. (#14076)

* Introduce hidden flag `data-columns-withhold-count`.

* Address Nishant's comment.

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

@@ -458,14 +458,28 @@ func (vs *Server) broadcastAndReceiveBlobs(ctx context.Context, sidecars []*ethp
 // broadcastAndReceiveDataColumns handles the broadcasting and reception of data columns sidecars.
 func (vs *Server) broadcastAndReceiveDataColumns(ctx context.Context, sidecars []*ethpb.DataColumnSidecar, root [fieldparams.RootLength]byte) error {
 	eg, _ := errgroup.WithContext(ctx)
+
+	dataColumnsWithholdCount := features.Get().DataColumnsWithholdCount
+
 	for i, sd := range sidecars {
 		// Copy the iteration instance to a local variable to give each go-routine its own copy to play with.
 		// See https://golang.org/doc/faq#closures_and_goroutines for more details.
-		colIdx := i
-		sidecar := sd
+		colIdx, sidecar := i, sd
+
 		eg.Go(func() error {
-			if err := vs.P2P.BroadcastDataColumn(ctx, uint64(colIdx)%params.BeaconConfig().DataColumnSidecarSubnetCount, sidecar); err != nil {
-				return errors.Wrap(err, "broadcast data column")
+			// Compute the subnet index based on the column index.
+			subnet := uint64(colIdx) % params.BeaconConfig().DataColumnSidecarSubnetCount
+
+			if colIdx < dataColumnsWithholdCount {
+				log.WithFields(logrus.Fields{
+					"root":            fmt.Sprintf("%#x", root),
+					"subnet":          subnet,
+					"dataColumnIndex": colIdx,
+				}).Warning("Withholding data column")
+			} else {
+				if err := vs.P2P.BroadcastDataColumn(ctx, subnet, sidecar); err != nil {
+					return errors.Wrap(err, "broadcast data column")
+				}
 			}
 
 			roDataColumn, err := blocks.NewRODataColumnWithRoot(sidecar, root)

config/features/config.go

@@ -77,6 +77,9 @@ type Flags struct {
 	// EnablePeerDAS enables running the node with the experimental data availability sampling scheme.
 	EnablePeerDAS bool
 
+	// DataColumnsWithholdCount specifies the likelihood of withholding a data column sidecar when proposing a block (percentage)
+	DataColumnsWithholdCount int
+
 	SaveInvalidBlock bool // SaveInvalidBlock saves invalid block to temp.
 	SaveInvalidBlob  bool // SaveInvalidBlob saves invalid blob to temp.
 
@@ -268,6 +271,11 @@ func ConfigureBeaconChain(ctx *cli.Context) error {
 		cfg.EnablePeerDAS = true
 	}
 
+	if ctx.IsSet(DataColumnsWithholdCount.Name) {
+		logEnabled(DataColumnsWithholdCount)
+		cfg.DataColumnsWithholdCount = ctx.Int(DataColumnsWithholdCount.Name)
+	}
+
 	cfg.AggregateIntervals = [3]time.Duration{aggregateFirstInterval.Value, aggregateSecondInterval.Value, aggregateThirdInterval.Value}
 	Init(cfg)
 	return nil

config/features/flags.go

@@ -166,6 +166,7 @@ var (
 		Name:  "enable-quic",
 		Usage: "Enables connection using the QUIC protocol for peers which support it.",
 	}
+	// EnablePeerDAS is a flag for enabling the peer data availability sampling.
 	EnableCommitteeAwarePacking = &cli.BoolFlag{
 		Name:  "enable-committee-aware-packing",
 		Usage: "Changes the attestation packing algorithm to one that is aware of attesting committees.",
@@ -174,6 +175,13 @@ var (
 		Name:  "peer-das",
 		Usage: "Enables Prysm to run with the experimental peer data availability sampling scheme.",
 	}
+	// DataColumnsWithholdCount is a flag for withholding data columns when proposing a block.
+	DataColumnsWithholdCount = &cli.IntFlag{
+		Name:   "data-columns-withhold-count",
+		Usage:  "Number of columns to withhold when proposing a block. DO NOT USE IN PRODUCTION.",
+		Value:  0,
+		Hidden: true,
+	}
 )
 
 // devModeFlags holds list of flags that are set when development mode is on.
@@ -233,6 +241,7 @@ var BeaconChainFlags = append(deprecatedBeaconFlags, append(deprecatedFlags, []c
 	EnableQUIC,
 	EnableCommitteeAwarePacking,
 	EnablePeerDAS,
+	DataColumnsWithholdCount,
 }...)...)
 
 // E2EBeaconChainFlags contains a list of the beacon chain feature flags to be tested in E2E.