Simplify DA check to avoid blob/block timing inconsistencies (#12882)

* collect unique idxs and avoid races

* fix init

* remove unused type (lint error)

---------

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

beacon-chain/blockchain/process_block.go

@@ -16,7 +16,6 @@ import (
 	forkchoicetypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/forkchoice/types"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v4/config/features"
-	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	consensusblocks "github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
@@ -552,21 +551,16 @@ func (s *Service) isDataAvailable(ctx context.Context, root [32]byte, signed int
 	if err != nil {
 		return errors.Wrap(err, "could not get KZG commitments")
 	}
-	existingBlobs := len(kzgCommitments)
-	if existingBlobs == 0 {
+	expected := len(kzgCommitments)
+	if expected == 0 {
 		return nil
 	}
 
 	// Read first from db in case we have the blobs
-	s.blobNotifier.Lock()
-	var nc *blobNotifierChan
-	var ok bool
-	nc, ok = s.blobNotifier.chanForRoot[root]
 	sidecars, err := s.cfg.BeaconDB.BlobSidecarsByRoot(ctx, root)
 	if err == nil {
-		if len(sidecars) >= existingBlobs {
-			delete(s.blobNotifier.chanForRoot, root)
-			s.blobNotifier.Unlock()
+		if len(sidecars) >= expected {
+			s.blobNotifiers.delete(root)
 			if err := kzg.IsDataAvailable(kzgCommitments, sidecars); err != nil {
 				return err
 			}
@@ -574,38 +568,29 @@ func (s *Service) isDataAvailable(ctx context.Context, root [32]byte, signed int
 			return nil
 		}
 	}
-	// Create the channel if it didn't exist already the index map will be
-	// created later anyway
-	if !ok {
-		nc = &blobNotifierChan{channel: make(chan struct{}, fieldparams.MaxBlobsPerBlock)}
-		s.blobNotifier.chanForRoot[root] = nc
+
+	found := map[uint64]struct{}{}
+	for _, sc := range sidecars {
+		found[sc.Index] = struct{}{}
 	}
-	// We have more commitments in the block than blobs in database
-	// We sync the channel indices with the sidecars
-	nc.indices = make(map[uint64]struct{})
-	for _, sidecar := range sidecars {
-		nc.indices[sidecar.Index] = struct{}{}
-	}
-	s.blobNotifier.Unlock()
-	channelWrites := len(sidecars)
+	nc := s.blobNotifiers.forRoot(root)
 	for {
 		select {
-		case <-nc.channel:
-			channelWrites++
-			if channelWrites == existingBlobs {
-				s.blobNotifier.Lock()
-				delete(s.blobNotifier.chanForRoot, root)
-				s.blobNotifier.Unlock()
-				sidecars, err := s.cfg.BeaconDB.BlobSidecarsByRoot(ctx, root)
-				if err != nil {
-					return errors.Wrap(err, "could not get blob sidecars")
-				}
-				if err := kzg.IsDataAvailable(kzgCommitments, sidecars); err != nil {
-					return err
-				}
-				logBlobSidecar(sidecars, t)
-				return nil
+		case idx := <-nc:
+			found[idx] = struct{}{}
+			if len(found) != expected {
+				continue
 			}
+			s.blobNotifiers.delete(root)
+			sidecars, err := s.cfg.BeaconDB.BlobSidecarsByRoot(ctx, root)
+			if err != nil {
+				return errors.Wrap(err, "could not get blob sidecars")
+			}
+			if err := kzg.IsDataAvailable(kzgCommitments, sidecars); err != nil {
+				return err
+			}
+			logBlobSidecar(sidecars, t)
+			return nil
 		case <-ctx.Done():
 			return errors.Wrap(ctx.Err(), "context deadline waiting for blob sidecars")
 		}

beacon-chain/blockchain/receive_blob.go

@@ -1,22 +1,7 @@
 package blockchain
 
-import fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
-
 // SendNewBlobEvent sends a message to the BlobNotifier channel that the blob
 // for the blocroot `root` is ready in the database
 func (s *Service) SendNewBlobEvent(root [32]byte, index uint64) {
-	s.blobNotifier.Lock()
-	nc, ok := s.blobNotifier.chanForRoot[root]
-	if !ok {
-		nc = &blobNotifierChan{indices: make(map[uint64]struct{}), channel: make(chan struct{}, fieldparams.MaxBlobsPerBlock)}
-		s.blobNotifier.chanForRoot[root] = nc
-	}
-	_, ok = nc.indices[index]
-	if ok {
-		s.blobNotifier.Unlock()
-		return
-	}
-	nc.indices[index] = struct{}{}
-	s.blobNotifier.Unlock()
-	nc.channel <- struct{}{}
+	s.blobNotifiers.forRoot(root) <- index
 }

beacon-chain/blockchain/service.go

@@ -32,6 +32,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state/stategen"
 	"github.com/prysmaticlabs/prysm/v4/config/features"
+	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
@@ -60,7 +61,7 @@ type Service struct {
 	clockSetter          startup.ClockSetter
 	clockWaiter          startup.ClockWaiter
 	syncComplete         chan struct{}
-	blobNotifier         *blobNotifier
+	blobNotifiers        *blobNotifierMap
 }
 
 // config options for the service.
@@ -89,14 +90,26 @@ type config struct {
 
 var ErrMissingClockSetter = errors.New("blockchain Service initialized without a startup.ClockSetter")
 
-type blobNotifierChan struct {
-	indices map[uint64]struct{}
-	channel chan struct{}
+type blobNotifierMap struct {
+	sync.RWMutex
+	notifiers map[[32]byte]chan uint64
 }
 
-type blobNotifier struct {
-	sync.RWMutex
-	chanForRoot map[[32]byte]*blobNotifierChan
+func (bn *blobNotifierMap) forRoot(root [32]byte) chan uint64 {
+	bn.Lock()
+	defer bn.Unlock()
+	c, ok := bn.notifiers[root]
+	if !ok {
+		c = make(chan uint64, fieldparams.MaxBlobsPerBlock)
+		bn.notifiers[root] = c
+	}
+	return c
+}
+
+func (bn *blobNotifierMap) delete(root [32]byte) {
+	bn.Lock()
+	defer bn.Unlock()
+	delete(bn.notifiers, root)
 }
 
 // NewService instantiates a new block service instance that will
@@ -110,8 +123,8 @@ func NewService(ctx context.Context, opts ...Option) (*Service, error) {
 		}
 	}
 	ctx, cancel := context.WithCancel(ctx)
-	bn := &blobNotifier{
-		chanForRoot: make(map[[32]byte]*blobNotifierChan),
+	bn := &blobNotifierMap{
+		notifiers: make(map[[32]byte]chan uint64),
 	}
 	srv := &Service{
 		ctx:                  ctx,
@@ -119,7 +132,7 @@ func NewService(ctx context.Context, opts ...Option) (*Service, error) {
 		boundaryRoots:        [][32]byte{},
 		checkpointStateCache: cache.NewCheckpointStateCache(),
 		initSyncBlocks:       make(map[[32]byte]interfaces.ReadOnlySignedBeaconBlock),
-		blobNotifier:         bn,
+		blobNotifiers:        bn,
 		cfg:                  &config{ProposerSlotIndexCache: cache.NewProposerPayloadIDsCache()},
 	}
 	for _, opt := range opts {
@@ -367,7 +380,7 @@ func (s *Service) startFromExecutionChain() error {
 				log.Debug("Context closed, exiting goroutine")
 				return
 			case err := <-stateSub.Err():
-				log.WithError(err).Error("Subscription to state notifier failed")
+				log.WithError(err).Error("Subscription to state forRoot failed")
 				return
 			}
 		}