blobreactor implementation

Author: fridrik01

beacon/blockchain/common.go

@@ -27,42 +27,45 @@ import (
 	"github.com/berachain/beacon-kit/consensus/cometbft/service/encoding"
 	"github.com/berachain/beacon-kit/da/types"
 	"github.com/berachain/beacon-kit/primitives/math"
+	cmtabci "github.com/cometbft/cometbft/abci/types"
 )
 
-func (s *Service) ParseBeaconBlock(req encoding.ABCIRequest) (
+func (s *Service) ParseProcessProposalRequest(req *cmtabci.ProcessProposalRequest) (
 	*ctypes.SignedBeaconBlock,
 	types.BlobSidecars,
 	error,
 ) {
-	if countTx := len(req.GetTxs()); countTx > MaxConsensusTxsCount {
-		return nil, nil, fmt.Errorf("max expected %d, got %d: %w",
-			MaxConsensusTxsCount, countTx,
-			ErrTooManyConsensusTxs,
-		)
+	blobConsensusEnabled := s.chainSpec.BlobConsensusEnableHeight() > 0 && req.Height >= s.chainSpec.BlobConsensusEnableHeight()
+
+	maxTxCount := MaxConsensusTxsCount
+	if blobConsensusEnabled {
+		maxTxCount = 1 // After BlobEnableHeight: only 1 tx expected (block), blobs in Blob field
+	}
+
+	if len(req.GetTxs()) > maxTxCount {
+		return nil, nil, fmt.Errorf("max expected %d txs, got %d", maxTxCount, len(req.GetTxs()))
 	}
 
 	forkVersion := s.chainSpec.ActiveForkVersionForTimestamp(math.U64(req.GetTime().Unix())) //#nosec: G115
-	// Decode signed block and sidecars.
-	signedBlk, sidecars, err := encoding.ExtractBlobsAndBlockFromRequest(
-		req,
-		BeaconBlockTxIndex,
-		BlobSidecarsTxIndex,
-		forkVersion,
-	)
+	signedBlk, err := encoding.UnmarshalBeaconBlockFromABCIRequest(req.GetTxs(), BeaconBlockTxIndex, forkVersion)
 	if err != nil {
 		return nil, nil, err
 	}
+
 	if signedBlk == nil {
-		s.logger.Warn(
-			"Aborting block verification - beacon block not found in proposal",
-		)
+		s.logger.Warn("Aborting block verification - beacon block not found in proposal")
 		return nil, nil, ErrNilBlk
 	}
-	if sidecars == nil {
-		s.logger.Warn(
-			"Aborting block verification - blob sidecars not found in proposal",
-		)
-		return nil, nil, ErrNilBlob
+
+	// Extract sidecars using the common helper
+	sidecars, err := encoding.ExtractBlobSidecarsFromRequest(
+		req.GetTxs(),
+		req.GetBlob(),
+		req.Height,
+		s.chainSpec.BlobConsensusEnableHeight(),
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to extract blob sidecars at height %d: %w", req.Height, err)
 	}
 
 	return signedBlk, sidecars, nil

beacon/blockchain/finalize_block.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	ctypes "github.com/berachain/beacon-kit/consensus-types/types"
+	"github.com/berachain/beacon-kit/consensus/cometbft/service/encoding"
 	"github.com/berachain/beacon-kit/consensus/types"
 	datypes "github.com/berachain/beacon-kit/da/types"
 	"github.com/berachain/beacon-kit/primitives/crypto"
@@ -39,12 +40,27 @@ import (
 func (s *Service) FinalizeBlock(
 	ctx sdk.Context,
 	req *cmtabci.FinalizeBlockRequest,
+	blobs datypes.BlobSidecars,
 ) (transition.ValidatorUpdates, error) {
+	maxTxCount := MaxConsensusTxsCount
+	if s.chainSpec.BlobConsensusEnableHeight() > 0 && req.Height >= s.chainSpec.BlobConsensusEnableHeight() {
+		maxTxCount = 1
+	}
+
 	// STEP 1: Decode block and blobs.
-	signedBlk, blobs, err := s.ParseBeaconBlock(req)
+	if countTx := len(req.GetTxs()); countTx > maxTxCount {
+		return nil, fmt.Errorf("invalid tx count in FinalizeBlock at height %d: expected max %d, got %d",
+			req.Height, maxTxCount, countTx)
+	}
+
+	forkVersion := s.chainSpec.ActiveForkVersionForTimestamp(math.U64(req.GetTime().Unix())) //#nosec: G115
+	signedBlk, err := encoding.UnmarshalBeaconBlockFromABCIRequest(
+		req.GetTxs(),
+		BeaconBlockTxIndex,
+		forkVersion,
+	)
 	if err != nil {
-		s.logger.Error("Failed to decode block and blobs", "error", err)
-		return nil, fmt.Errorf("failed to decode block and blobs: %w", err)
+		return nil, fmt.Errorf("failed to decode block at height %d with fork version %s: %w", req.Height, forkVersion, err)
 	}
 	blk := signedBlk.GetBeaconBlock()
 	st := s.storageBackend.StateFromContext(ctx)
@@ -73,43 +89,95 @@ func (s *Service) FinalizeBlock(
 	consensusBlk := types.NewConsensusBlock(blk, req.GetProposerAddress(), req.GetTime())
 	valUpdates, err := s.finalizeBeaconBlock(ctx, st, consensusBlk)
 	if err != nil {
-		s.logger.Error("Failed to process verified beacon block",
-			"error", err,
-		)
-		return nil, err
+		return nil, fmt.Errorf("failed finalizing beacon block at height %d: %w", req.Height, err)
 	}
 
 	// STEP 4: Post Finalizations cleanups.
 	return valUpdates, s.PostFinalizeBlockOps(ctx, blk)
 }
 
+//nolint:gocognit // ok for now
 func (s *Service) FinalizeSidecars(
 	ctx sdk.Context,
 	syncingToHeight int64,
 	blk *ctypes.BeaconBlock,
 	blobs datypes.BlobSidecars,
 ) error {
+	// Check the block to see if there should be blobs
+	expectedBlobs := len(blk.GetBody().GetBlobKzgCommitments())
+	if expectedBlobs == 0 {
+		return nil // No blobs expected for this block
+	}
+
+	processBlobsFunc := func(blobs datypes.BlobSidecars) error {
+		// Process the blobs which saves them to storage
+		if err := s.blobProcessor.ProcessSidecars(s.storageBackend.AvailabilityStore(), blobs); err != nil {
+			return fmt.Errorf("failed to process %d blob sidecars for slot %d: %w", len(blobs), blk.GetSlot(), err)
+		}
+
+		// Final verification
+		if !s.storageBackend.AvailabilityStore().IsDataAvailable(ctx, blk.GetSlot(), blk.GetBody()) {
+			return fmt.Errorf("data not available after processing blobs for slot %d: %w", blk.GetSlot(), ErrDataNotAvailable)
+		}
+
+		return nil
+	}
+
 	// SyncingToHeight is always the tip of the chain both during sync and when
 	// caught up. We don't need to process sidecars unless they are within DA period.
 	//
 	//#nosec: G115 // SyncingToHeight will never be negative.
+	//nolint:nestif // ok for now
 	if s.chainSpec.WithinDAPeriod(blk.GetSlot(), math.Slot(syncingToHeight)) {
-		err := s.blobProcessor.ProcessSidecars(
-			s.storageBackend.AvailabilityStore(),
-			blobs,
-		)
+		if s.storageBackend.AvailabilityStore().IsDataAvailable(ctx, blk.GetSlot(), blk.GetBody()) {
+			return nil // Data already available
+		}
+
+		// if we are before blob enable height then blobs should be in storage already
+		if s.chainSpec.BlobConsensusEnableHeight() <= 0 || int64(blk.GetSlot()) < s.chainSpec.BlobConsensusEnableHeight() {
+			return processBlobsFunc(blobs)
+		}
+
+		// If blobs were passed in (normal consensus mode), use them
+		if len(blobs) > 0 {
+			return processBlobsFunc(blobs)
+		}
+
+		// Otherwise we need to fetch the blobs from peers (sync mode)
+		if s.blobRequester == nil {
+			return fmt.Errorf("blob requester not initialized for slot %d after BlobEnableHeight %d",
+				blk.GetSlot(), s.chainSpec.BlobConsensusEnableHeight())
+		}
+
+		// Check if we're at the current consensus height (switching from sync to consensus). In this case, we're finalizing a block
+		// that was just agreed upon in consensus, and blob fetching would fail since peers are still processing the same height
+		if int64(blk.GetSlot()) >= syncingToHeight {
+			s.logger.Warn("At consensus height during mode switch, skipping blob fetch",
+				"slot", blk.GetSlot(),
+				"syncingToHeight", syncingToHeight,
+				"expected_blobs", expectedBlobs)
+			return nil
+		}
+
+		s.logger.Info("Fetching missing blobs during sync", "slot", blk.GetSlot(), "expected_blobs", expectedBlobs)
+		now := time.Now()
+		fetchedBlobs, err := s.blobRequester.RequestBlobs(blk.GetSlot().Unwrap())
 		if err != nil {
-			s.logger.Error("Failed to process blob sidecars", "error", err)
-			return fmt.Errorf("failed to process blob sidecars: %w", err)
+			return fmt.Errorf("failed to fetch blobs, slot: %d: %w", blk.GetSlot(), err)
 		}
 
-		// Ensure we can access the data using the commitments from the block.
-		if !s.storageBackend.AvailabilityStore().IsDataAvailable(
-			ctx, blk.GetSlot(), blk.GetBody(),
-		) {
-			return ErrDataNotAvailable
+		// Validate blobs fetched from peers before processing
+		err = s.blobProcessor.VerifySidecars(ctx, fetchedBlobs, blk.GetHeader(), blk.GetBody().GetBlobKzgCommitments())
+		if err != nil {
+			return fmt.Errorf("failed to verify fetched blobs from peers at slot %d: %w", blk.GetSlot(), err)
 		}
-		return nil
+
+		s.logger.Info("Successfully fetched and verified blobs",
+			"slot", blk.GetSlot(),
+			"count", len(fetchedBlobs),
+			"elapsed", time.Since(now).String())
+
+		return processBlobsFunc(fetchedBlobs)
 	}
 
 	// Here outside Data Availability window. Just log if needed
@@ -141,6 +209,11 @@ func (s *Service) PostFinalizeBlockOps(ctx sdk.Context, blk *ctypes.BeaconBlock)
 		return err
 	}
 
+	// Update the BlobReactor with the current head slot so it can advertise to peers
+	if s.blobRequester != nil {
+		s.blobRequester.SetHeadSlot(slot.Unwrap())
+	}
+
 	// Prune the availability and deposit store.
 	if err := s.processPruning(ctx, blk); err != nil {
 		s.logger.Error("failed to processPruning", "error", err)

beacon/blockchain/interfaces.go

@@ -26,8 +26,8 @@ import (
 
 	"github.com/berachain/beacon-kit/chain"
 	ctypes "github.com/berachain/beacon-kit/consensus-types/types"
+	"github.com/berachain/beacon-kit/consensus/cometbft/service/blobreactor"
 	"github.com/berachain/beacon-kit/consensus/cometbft/service/delay"
-	"github.com/berachain/beacon-kit/consensus/cometbft/service/encoding"
 	dastore "github.com/berachain/beacon-kit/da/store"
 	datypes "github.com/berachain/beacon-kit/da/types"
 	engineprimitives "github.com/berachain/beacon-kit/engine-primitives/engine-primitives"
@@ -45,6 +45,17 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
+// BlobRequester is the interface for requesting blobs from peers.
+type BlobRequester interface {
+	// RequestBlobs fetches all blobs for a given slot from peers.
+	// Returns all blob sidecars for the slot, or an error if none could be retrieved.
+	RequestBlobs(slot uint64) ([]*datypes.BlobSidecar, error)
+
+	// SetHeadSlot updates the reactor's view of the current blockchain head slot.
+	// Called by the blockchain service after processing each block.
+	SetHeadSlot(slot uint64)
+}
+
 // ExecutionEngine is the interface for the execution engine.
 type ExecutionEngine interface {
 	// NotifyNewPayload notifies the execution client of new payload.
@@ -135,7 +146,7 @@ type BlockchainI interface {
 		context.Context,
 		[]byte,
 	) (transition.ValidatorUpdates, error)
-	ParseBeaconBlock(req encoding.ABCIRequest) (
+	ParseProcessProposalRequest(*cmtabci.ProcessProposalRequest) (
 		*ctypes.SignedBeaconBlock,
 		datypes.BlobSidecars,
 		error,
@@ -154,6 +165,7 @@ type BlockchainI interface {
 	FinalizeBlock(
 		sdk.Context,
 		*cmtabci.FinalizeBlockRequest,
+		datypes.BlobSidecars,
 	) (transition.ValidatorUpdates, error)
 	PostFinalizeBlockOps(
 		sdk.Context,
@@ -189,6 +201,7 @@ type ServiceChainSpec interface {
 	chain.ForkSpec
 	chain.ForkVersionSpec
 	delay.ConfigGetter
+	blobreactor.ConfigGetter
 
 	EpochsPerHistoricalVector() uint64
 	SlotToEpoch(slot math.Slot) math.Epoch

beacon/blockchain/payload_test.go

@@ -277,6 +277,7 @@ func setupOptimisticPayloadTests(t *testing.T, cs chain.Spec, optimisticPayloadB
 	chain := blockchain.NewService(
 		sb,
 		nil, // blockchain.BlobProcessor unused in this test
+		nil, // blockchain.BlobRequester unused in this test
 		nil, // deposit.Contract unused in this test
 		logger,
 		cs,

beacon/blockchain/process_proposal.go

@@ -45,7 +45,6 @@ import (
 )
 
 const (
-	// BeaconBlockTxIndex represents the index of the beacon block transaction.
 	// It is the first transaction in the tx list.
 	BeaconBlockTxIndex uint = iota
 	// BlobSidecarsTxIndex represents the index of the blob sidecar transaction.
@@ -62,7 +61,7 @@ func (s *Service) ProcessProposal(
 	req *cmtabci.ProcessProposalRequest,
 	thisNodeAddress []byte,
 ) (transition.ValidatorUpdates, error) {
-	signedBlk, sidecars, err := s.ParseBeaconBlock(req)
+	signedBlk, sidecars, err := s.ParseProcessProposalRequest(req)
 	if err != nil {
 		s.logger.Error("Failed to decode block and blobs", "error", err)
 		return nil, fmt.Errorf("failed to decode block and blobs: %w", err)

beacon/blockchain/service.go

@@ -35,6 +35,8 @@ type Service struct {
 	storageBackend StorageBackend
 	// blobProcessor is used for processing sidecars.
 	blobProcessor BlobProcessor
+	// blobRequester is used for requesting missing blobs from peers during sync.
+	blobRequester BlobRequester
 	// depositContract is the contract interface for interacting with the
 	// deposit contract.
 	depositContract deposit.Contract
@@ -70,6 +72,7 @@ type Service struct {
 func NewService(
 	storageBackend StorageBackend,
 	blobProcessor BlobProcessor,
+	blobRequester BlobRequester,
 	depositContract deposit.Contract,
 	logger log.Logger,
 	chainSpec ServiceChainSpec,
@@ -82,6 +85,7 @@ func NewService(
 	return &Service{
 		storageBackend:          storageBackend,
 		blobProcessor:           blobProcessor,
+		blobRequester:           blobRequester,
 		depositContract:         depositContract,
 		eth1FollowDistance:      math.U64(chainSpec.Eth1FollowDistance()),
 		failedBlocks:            make(map[math.Slot]struct{}),

chain/data.go

@@ -21,6 +21,7 @@
 package chain
 
 import (
+	"github.com/berachain/beacon-kit/consensus/cometbft/service/blobreactor"
 	"github.com/berachain/beacon-kit/consensus/cometbft/service/delay"
 	"github.com/berachain/beacon-kit/primitives/common"
 )
@@ -29,6 +30,7 @@ import (
 // `mapstructure` tag are required.
 type SpecData struct {
 	delay.Config `mapstructure:"block-delay-configuration"`
+	BlobConfig   blobreactor.Config `mapstructure:"blob-reactor-configuration"`
 
 	// Gwei value constants.
 	//