Fix data columns sampling (#14263)

* Fix the obvious...

* Data columns sampling: Modify logging.

* `waitForChainStart`: Set it threadsafe - Do only wait once.

* Sampling: Wait for chain start before running the sampling.

Reason: `newDataColumnSampler1D` needs `s.ctxMap`.
`s.ctxMap` is only set when chain is started.

Previously `waitForChainStart` was only called in `s.registerHandlers`, it self called in a go-routine.

==> We had a race condition here: Sometimes `newDataColumnSampler1D` were called once `s.ctxMap` were set, sometimes not.

* Adresse Nishant's comments.

* Sampling: Improve logging.

* `waitForChainStart`: Remove `chainIsStarted` check.

beacon-chain/sync/data_columns_sampling.go

@@ -170,8 +170,6 @@ func (d *dataColumnSampler1D) refreshPeerInfo() {
 		}
 	}
 
-	log.WithField("columnFromPeer", d.columnFromPeer).Debug("Peer info refreshed")
-
 	columnWithNoPeers := make([]uint64, 0)
 	for column, peers := range d.peerFromColumn {
 		if len(peers) == 0 {
@@ -228,7 +226,7 @@ func (d *dataColumnSampler1D) handleStateNotification(ctx context.Context, event
 		return
 	}
 
-	if coreTime.PeerDASIsActive(data.Slot) {
+	if !coreTime.PeerDASIsActive(data.Slot) {
 		// We do not trigger sampling if peerDAS is not active yet.
 		return
 	}
@@ -249,22 +247,12 @@ func (d *dataColumnSampler1D) handleStateNotification(ctx context.Context, event
 	// Randomize columns for sample selection.
 	randomizedColumns := randomizeColumns(d.nonCustodyColumns)
 	samplesCount := min(params.BeaconConfig().SamplesPerSlot, uint64(len(d.nonCustodyColumns))-params.BeaconConfig().NumberOfColumns/2)
-	ok, _, err = d.incrementalDAS(ctx, data.BlockRoot, randomizedColumns, samplesCount)
+
+	// TODO: Use the first output of `incrementalDAS` as input of the fork choice rule.
+	_, _, err = d.incrementalDAS(ctx, data.BlockRoot, randomizedColumns, samplesCount)
 	if err != nil {
 		log.WithError(err).Error("Failed to run incremental DAS")
 	}
-
-	if ok {
-		log.WithFields(logrus.Fields{
-			"root":    fmt.Sprintf("%#x", data.BlockRoot),
-			"columns": randomizedColumns,
-		}).Debug("Data column sampling successful")
-	} else {
-		log.WithFields(logrus.Fields{
-			"root":    fmt.Sprintf("%#x", data.BlockRoot),
-			"columns": randomizedColumns,
-		}).Warning("Data column sampling failed")
-	}
 }
 
 // incrementalDAS samples data columns from active peers using incremental DAS.
@@ -280,17 +268,28 @@ func (d *dataColumnSampler1D) incrementalDAS(
 	firstColumnToSample, extendedSampleCount := uint64(0), peerdas.ExtendedSampleCount(sampleCount, allowedFailures)
 	roundSummaries := make([]roundSummary, 0, 1) // We optimistically allocate only one round summary.
 
+	start := time.Now()
+
 	for round := 1; ; /*No exit condition */ round++ {
 		if extendedSampleCount > uint64(len(columns)) {
 			// We already tried to sample all possible columns, this is the unhappy path.
-			log.WithField("root", fmt.Sprintf("%#x", root)).Warning("Some columns are still missing after sampling all possible columns")
+			log.WithFields(logrus.Fields{
+				"root":  fmt.Sprintf("%#x", root),
+				"round": round - 1,
+			}).Warning("Some columns are still missing after trying to sample all possible columns")
 			return false, roundSummaries, nil
 		}
 
 		// Get the columns to sample for this round.
 		columnsToSample := columns[firstColumnToSample:extendedSampleCount]
 		columnsToSampleCount := extendedSampleCount - firstColumnToSample
 
+		log.WithFields(logrus.Fields{
+			"root":    fmt.Sprintf("%#x", root),
+			"columns": columnsToSample,
+			"round":   round,
+		}).Debug("Start data columns sampling")
+
 		// Sample data columns from peers in parallel.
 		retrievedSamples := d.sampleDataColumns(ctx, root, columnsToSample)
 
@@ -311,7 +310,8 @@ func (d *dataColumnSampler1D) incrementalDAS(
 			// All columns were correctly sampled, this is the happy path.
 			log.WithFields(logrus.Fields{
 				"root":         fmt.Sprintf("%#x", root),
-				"roundsNeeded": round,
+				"neededRounds": round,
+				"duration":     time.Since(start),
 			}).Debug("All columns were successfully sampled")
 			return true, roundSummaries, nil
 		}
@@ -429,14 +429,14 @@ func (d *dataColumnSampler1D) sampleDataColumnsFromPeer(
 			"peerID":           pid,
 			"root":             fmt.Sprintf("%#x", root),
 			"requestedColumns": sortedSliceFromMap(requestedColumns),
-		}).Debug("All requested columns were successfully sampled from peer")
+		}).Debug("Sampled columns from peer successfully")
 	} else {
 		log.WithFields(logrus.Fields{
 			"peerID":           pid,
 			"root":             fmt.Sprintf("%#x", root),
 			"requestedColumns": sortedSliceFromMap(requestedColumns),
 			"retrievedColumns": sortedSliceFromMap(retrievedColumns),
-		}).Debug("Some requested columns were not sampled from peer")
+		}).Debug("Sampled columns from peer with some errors")
 	}
 
 	return retrievedColumns

beacon-chain/sync/rpc_data_column_sidecars_by_root.go

@@ -94,7 +94,7 @@ func (s *Service) dataColumnSidecarByRootRPCHandler(ctx context.Context, msg int
 		"requested":      requestedColumnsList,
 		"custodiedCount": len(custodiedColumnsList),
 		"requestedCount": len(requestedColumnsList),
-	}).Debug("Received data column sidecar by root request")
+	}).Debug("Data column sidecar by root request received")
 
 	// Subscribe to the data column feed.
 	rootIndexChan := make(chan filesystem.RootIndexPair)

beacon-chain/sync/rpc_send_request.go

@@ -232,7 +232,6 @@ func SendDataColumnSidecarByRoot(
 	}
 
 	// Send the request to the peer.
-	log.WithField("topic", topic).Debug("Sending data column sidecar request")
 	stream, err := p2pApi.Send(ctx, req, topic, pid)
 	if err != nil {
 		return nil, errors.Wrap(err, "send")

beacon-chain/sync/service.go

@@ -239,7 +239,7 @@ func (s *Service) Start() {
 	s.newColumnProposerVerifier = v.VerifyProposer
 
 	go s.verifierRoutine()
-	go s.registerHandlers()
+	go s.startTasksPostInitialSync()
 
 	s.cfg.p2p.AddConnectionHandler(s.reValidatePeer, s.sendGoodbye)
 	s.cfg.p2p.AddDisconnectionHandler(func(_ context.Context, _ peer.ID) error {
@@ -254,12 +254,6 @@ func (s *Service) Start() {
 
 	// Update sync metrics.
 	async.RunEvery(s.ctx, syncMetricsInterval, s.updateMetrics)
-
-	// Run data column sampling
-	if params.PeerDASEnabled() {
-		s.sampler = newDataColumnSampler1D(s.cfg.p2p, s.cfg.clock, s.ctxMap, s.cfg.stateNotifier)
-		go s.sampler.Run(s.ctx)
-	}
 }
 
 // Stop the regular sync service.
@@ -337,23 +331,37 @@ func (s *Service) waitForChainStart() {
 	s.markForChainStart()
 }
 
-func (s *Service) registerHandlers() {
+func (s *Service) startTasksPostInitialSync() {
+	// Wait for the chain to start.
 	s.waitForChainStart()
+
 	select {
 	case <-s.initialSyncComplete:
-		// Register respective pubsub handlers at state synced event.
-		digest, err := s.currentForkDigest()
+		// Compute the current epoch.
+		currentSlot := slots.CurrentSlot(uint64(s.cfg.clock.GenesisTime().Unix()))
+		currentEpoch := slots.ToEpoch(currentSlot)
+
+		// Compute the current fork forkDigest.
+		forkDigest, err := s.currentForkDigest()
 		if err != nil {
 			log.WithError(err).Error("Could not retrieve current fork digest")
 			return
 		}
-		currentEpoch := slots.ToEpoch(slots.CurrentSlot(uint64(s.cfg.clock.GenesisTime().Unix())))
-		s.registerSubscribers(currentEpoch, digest)
+
+		// Register respective pubsub handlers at state synced event.
+		s.registerSubscribers(currentEpoch, forkDigest)
+
+		// Start the fork watcher.
 		go s.forkWatcher()
-		return
+
+		// Start data columns sampling if peerDAS is enabled.
+		if params.PeerDASEnabled() {
+			s.sampler = newDataColumnSampler1D(s.cfg.p2p, s.cfg.clock, s.ctxMap, s.cfg.stateNotifier)
+			go s.sampler.Run(s.ctx)
+		}
+
 	case <-s.ctx.Done():
 		log.Debug("Context closed, exiting goroutine")
-		return
 	}
 }
 

beacon-chain/sync/service_test.go

@@ -62,7 +62,7 @@ func TestSyncHandlers_WaitToSync(t *testing.T) {
 	}
 
 	topic := "/eth2/%x/beacon_block"
-	go r.registerHandlers()
+	go r.startTasksPostInitialSync()
 	time.Sleep(100 * time.Millisecond)
 
 	var vr [32]byte
@@ -143,7 +143,7 @@ func TestSyncHandlers_WaitTillSynced(t *testing.T) {
 
 	syncCompleteCh := make(chan bool)
 	go func() {
-		r.registerHandlers()
+		r.startTasksPostInitialSync()
 		syncCompleteCh <- true
 	}()
 
@@ -200,7 +200,7 @@ func TestSyncService_StopCleanly(t *testing.T) {
 		initialSyncComplete: make(chan struct{}),
 	}
 
-	go r.registerHandlers()
+	go r.startTasksPostInitialSync()
 	var vr [32]byte
 	require.NoError(t, gs.SetClock(startup.NewClock(time.Now(), vr)))
 	r.waitForChainStart()