bug(blockchain): make block finalization wait on sidecar availability

mod/beacon/blockchain/process.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/berachain/beacon-kit/mod/primitives/pkg/async"
 	"github.com/berachain/beacon-kit/mod/primitives/pkg/transition"
+	"golang.org/x/sync/errgroup"
 )
 
 // ProcessGenesisData processes the genesis state and initializes the beacon
@@ -63,25 +64,26 @@ func (s *Service[
 		return nil, err
 	}
 
-	// If the blobs needed to process the block are not available, we
-	// return an error. It is safe to use the slot off of the beacon block
-	// since it has been verified as correct already.
-	if !s.storageBackend.AvailabilityStore().IsDataAvailable(
-		ctx, blk.GetSlot(), blk.GetBody(),
-	) {
-		return nil, ErrDataNotAvailable
+	// To complete block finalization, we wait for the associated blob
+	// to be persisted and verify it's available. Once that is done we
+	// complete block finalization.
+	g, _ := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		return s.verifyFinalBlobAvailability(ctx, blk)
+	})
+
+	// Wait for the sidecar to be finalized and its availability checked
+	if err = g.Wait(); err != nil {
+		return nil, err
 	}
 
 	// If required, we want to forkchoice at the end of post
 	// block processing.
 	// TODO: this is hood as fuck.
 	// We won't send an fcu if the block is bad, should be addressed
 	// via ticker later.
-	if err = s.dispatcher.Publish(
-		async.NewEvent(
-			ctx, async.BeaconBlockFinalized, blk,
-		),
-	); err != nil {
+	event := async.NewEvent(ctx, async.BeaconBlockFinalized, blk)
+	if err = s.dispatcher.Publish(event); err != nil {
 		return nil, err
 	}
 

mod/beacon/blockchain/service.go

@@ -80,6 +80,12 @@ type Service[
 	// forceStartupSyncOnce is used to force a sync of the startup head.
 	forceStartupSyncOnce *sync.Once
 
+	// blobFinalized is used to verify blob sidecar availability upon
+	// block finalization
+	blobFinalized chan struct{}
+
+	// subBlobFinalized is a channel holding BlobSidecarsFinalized events.
+	subBlobFinalized chan async.Event[struct{}]
 	// subFinalBlkReceived is a channel holding FinalBeaconBlockReceived events.
 	subFinalBlkReceived chan async.Event[BeaconBlockT]
 	// subBlockReceived is a channel holding BeaconBlockReceived events.
@@ -142,6 +148,8 @@ func NewService[
 		metrics:                 newChainMetrics(telemetrySink),
 		optimisticPayloadBuilds: optimisticPayloadBuilds,
 		forceStartupSyncOnce:    new(sync.Once),
+		blobFinalized:           make(chan struct{}),
+		subBlobFinalized:        make(chan async.Event[struct{}]),
 		subFinalBlkReceived:     make(chan async.Event[BeaconBlockT]),
 		subBlockReceived:        make(chan async.Event[BeaconBlockT]),
 		subGenDataReceived:      make(chan async.Event[GenesisT]),
@@ -179,6 +187,12 @@ func (s *Service[
 		return err
 	}
 
+	if err := s.dispatcher.Subscribe(
+		async.BlobSidecarsFinalized, s.subBlobFinalized,
+	); err != nil {
+		return err
+	}
+
 	// start the main event loop to listen and handle events.
 	go s.eventLoop(ctx)
 	return nil
@@ -198,6 +212,8 @@ func (s *Service[
 			s.handleBeaconBlockReceived(event)
 		case event := <-s.subFinalBlkReceived:
 			s.handleBeaconBlockFinalization(event)
+		case event := <-s.subBlobFinalized:
+			s.blobFinalized <- event.Data()
 		}
 	}
 }
@@ -313,3 +329,27 @@ func (s *Service[
 		)
 	}
 }
+
+// Given the block blk, ready for finalization, verifyFinalBlobAvailability
+// waits for its blob sidecar to be available and verify it.
+func (s *Service[
+	_, BeaconBlockT, _, _, _, _, _, _, _, _,
+]) verifyFinalBlobAvailability(ctx context.Context, blk BeaconBlockT) error {
+	// wait for blob sidecar to be finalized
+	select {
+	case <-s.blobFinalized:
+		// Proceed with verification
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	// If the blobs needed to process the block are not available, we
+	// return an error. It is safe to use the slot off of the beacon block
+	// since it has been verified as correct already.
+	if !s.storageBackend.AvailabilityStore().IsDataAvailable(
+		ctx, blk.GetSlot(), blk.GetBody(),
+	) {
+		return ErrDataNotAvailable
+	}
+	return nil
+}

mod/da/pkg/da/service.go

@@ -132,8 +132,16 @@ func (s *Service[_, BlobSidecarsT]) handleFinalSidecarsReceived(
 	if err := s.processSidecars(msg.Context(), msg.Data()); err != nil {
 		s.logger.Error(
 			"Failed to process blob sidecars",
-			"error",
-			err,
+			"error", err,
+		)
+		return
+	}
+
+	event := async.NewEvent(msg.Context(), async.BlobSidecarsFinalized, struct{}{})
+	if err := s.dispatcher.Publish(event); err != nil {
+		s.logger.Error(
+			"Failed to publish blob finalized event",
+			"error", err,
 		)
 	}
 }

mod/node-core/pkg/components/dispatcher.go

@@ -57,6 +57,7 @@ func ProvideDispatcher[
 		dp.WithEvent[async.Event[BlobSidecarsT]](async.SidecarsVerified),
 		dp.WithEvent[async.Event[BeaconBlockT]](async.FinalBeaconBlockReceived),
 		dp.WithEvent[async.Event[BlobSidecarsT]](async.FinalSidecarsReceived),
+		dp.WithEvent[async.Event[struct{}]](async.BlobSidecarsFinalized),
 		dp.WithEvent[ValidatorUpdateEvent](
 			async.FinalValidatorUpdatesProcessed,
 		),

mod/primitives/pkg/async/id.go

@@ -46,4 +46,5 @@ const (
 	FinalSidecarsReceived          = "final-blob-sidecars-received"
 	FinalValidatorUpdatesProcessed = "final-validator-updates"
 	BeaconBlockFinalized           = "beacon-block-finalized"
+	BlobSidecarsFinalized          = "blob-sidecars-finalized"
 )