EIP-7732 support (ePBS)

blockdb/blockdb.go

@@ -45,17 +45,19 @@ func (db *BlockDb) Close() error {
 	return db.engine.Close()
 }
 
-func (db *BlockDb) GetBlock(ctx context.Context, slot uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error)) (*types.BlockData, error) {
-	return db.engine.GetBlock(ctx, slot, root, parseBlock)
+func (db *BlockDb) GetBlock(ctx context.Context, slot uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error), parsePayload func(uint64, []byte) (interface{}, error)) (*types.BlockData, error) {
+	return db.engine.GetBlock(ctx, slot, root, parseBlock, parsePayload)
 }
 
-func (db *BlockDb) AddBlock(ctx context.Context, slot uint64, root []byte, header_ver uint64, header_data []byte, body_ver uint64, body_data []byte) (bool, error) {
+func (db *BlockDb) AddBlock(ctx context.Context, slot uint64, root []byte, header_ver uint64, header_data []byte, body_ver uint64, body_data []byte, payload_ver uint64, payload_data []byte) (bool, error) {
 	return db.engine.AddBlock(ctx, slot, root, func() (*types.BlockData, error) {
 		return &types.BlockData{
-			HeaderVersion: header_ver,
-			HeaderData:    header_data,
-			BodyVersion:   body_ver,
-			BodyData:      body_data,
+			HeaderVersion:  header_ver,
+			HeaderData:     header_data,
+			BodyVersion:    body_ver,
+			BodyData:       body_data,
+			PayloadVersion: payload_ver,
+			PayloadData:    payload_data,
 		}, nil
 	})
 }

blockdb/pebble/pebble.go

@@ -14,8 +14,9 @@ const (
 )
 
 const (
-	BlockTypeHeader uint16 = 1
-	BlockTypeBody   uint16 = 2
+	BlockTypeHeader  uint16 = 1
+	BlockTypeBody    uint16 = 2
+	BlockTypePayload uint16 = 3
 )
 
 type PebbleEngine struct {
@@ -97,7 +98,34 @@ func (e *PebbleEngine) getBlockBody(root []byte, parser func(uint64, []byte) (in
 	return body, version, nil
 }
 
-func (e *PebbleEngine) GetBlock(_ context.Context, _ uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error)) (*types.BlockData, error) {
+func (e *PebbleEngine) getBlockPayload(root []byte, parser func(uint64, []byte) (interface{}, error)) (interface{}, uint64, error) {
+	key := make([]byte, 2+len(root)+2)
+	binary.BigEndian.PutUint16(key[:2], KeyNamespaceBlock)
+	copy(key[2:], root)
+	binary.BigEndian.PutUint16(key[2+len(root):], BlockTypePayload)
+
+	res, closer, err := e.db.Get(key)
+	if err != nil && err != pebble.ErrNotFound {
+		return nil, 0, err
+	}
+	defer closer.Close()
+
+	if err == pebble.ErrNotFound || len(res) == 0 {
+		return nil, 0, nil
+	}
+
+	version := binary.BigEndian.Uint64(res[:8])
+	block := res[8:]
+
+	body, err := parser(version, block)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return body, version, nil
+}
+
+func (e *PebbleEngine) GetBlock(_ context.Context, _ uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error), parsePayload func(uint64, []byte) (interface{}, error)) (*types.BlockData, error) {
 	header, header_ver, err := e.getBlockHeader(root)
 	if err != nil {
 		return nil, err
@@ -124,6 +152,14 @@ func (e *PebbleEngine) GetBlock(_ context.Context, _ uint64, root []byte, parseB
 	blockData.Body = body
 	blockData.BodyVersion = body_ver
 
+	payload, payload_ver, err := e.getBlockPayload(root, parsePayload)
+	if err != nil {
+		return nil, err
+	}
+
+	blockData.Payload = payload
+	blockData.PayloadVersion = payload_ver
+
 	return blockData, nil
 }
 
@@ -157,6 +193,19 @@ func (e *PebbleEngine) addBlockBody(root []byte, version uint64, block []byte) e
 	return e.db.Set(key, data, nil)
 }
 
+func (e *PebbleEngine) addBlockPayload(root []byte, version uint64, payload []byte) error {
+	key := make([]byte, 2+len(root)+2)
+	binary.BigEndian.PutUint16(key[:2], KeyNamespaceBlock)
+	copy(key[2:], root)
+	binary.BigEndian.PutUint16(key[2+len(root):], BlockTypePayload)
+
+	data := make([]byte, 8+len(payload))
+	binary.BigEndian.PutUint64(data[:8], version)
+	copy(data[8:], payload)
+
+	return e.db.Set(key, data, nil)
+}
+
 func (e *PebbleEngine) AddBlock(_ context.Context, _ uint64, root []byte, dataCb func() (*types.BlockData, error)) (bool, error) {
 	key := make([]byte, 2+len(root)+2)
 	binary.BigEndian.PutUint16(key[:2], KeyNamespaceBlock)
@@ -182,5 +231,12 @@ func (e *PebbleEngine) AddBlock(_ context.Context, _ uint64, root []byte, dataCb
 		return false, err
 	}
 
+	if blockData.PayloadVersion != 0 {
+		err = e.addBlockPayload(root, blockData.PayloadVersion, blockData.PayloadData)
+		if err != nil {
+			return false, err
+		}
+	}
+
 	return true, nil
 }

blockdb/s3/s3store.go

@@ -60,10 +60,12 @@ func (e *S3Engine) getObjectKey(root []byte, slot uint64) string {
 }
 
 type objectMetadata struct {
-	objVersion   uint32
-	headerLength uint32
-	bodyVersion  uint32
-	bodyLength   uint32
+	objVersion     uint32
+	headerLength   uint32
+	bodyVersion    uint32
+	bodyLength     uint32
+	payloadVersion uint32
+	payloadLength  uint32
 }
 
 func (e *S3Engine) readObjectMetadata(data []byte) (*objectMetadata, int, error) {
@@ -78,6 +80,13 @@ func (e *S3Engine) readObjectMetadata(data []byte) (*objectMetadata, int, error)
 		metadata.bodyVersion = binary.BigEndian.Uint32(data[8:12])
 		metadata.bodyLength = binary.BigEndian.Uint32(data[12:16])
 		metadataLength += 12
+	case 2:
+		metadata.headerLength = binary.BigEndian.Uint32(data[4:8])
+		metadata.bodyVersion = binary.BigEndian.Uint32(data[8:12])
+		metadata.bodyLength = binary.BigEndian.Uint32(data[12:16])
+		metadata.payloadVersion = binary.BigEndian.Uint32(data[16:20])
+		metadata.payloadLength = binary.BigEndian.Uint32(data[20:24])
+		metadataLength += 20
 	}
 
 	return metadata, metadataLength, nil
@@ -92,12 +101,18 @@ func (e *S3Engine) writeObjectMetadata(metadata *objectMetadata) []byte {
 		data = binary.BigEndian.AppendUint32(data, metadata.headerLength)
 		data = binary.BigEndian.AppendUint32(data, metadata.bodyVersion)
 		data = binary.BigEndian.AppendUint32(data, metadata.bodyLength)
+	case 2:
+		data = binary.BigEndian.AppendUint32(data, metadata.headerLength)
+		data = binary.BigEndian.AppendUint32(data, metadata.bodyVersion)
+		data = binary.BigEndian.AppendUint32(data, metadata.bodyLength)
+		data = binary.BigEndian.AppendUint32(data, metadata.payloadVersion)
+		data = binary.BigEndian.AppendUint32(data, metadata.payloadLength)
 	}
 
 	return data
 }
 
-func (e *S3Engine) GetBlock(ctx context.Context, slot uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error)) (*types.BlockData, error) {
+func (e *S3Engine) GetBlock(ctx context.Context, slot uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error), parsePayload func(uint64, []byte) (interface{}, error)) (*types.BlockData, error) {
 	key := e.getObjectKey(root, slot)
 
 	obj, err := e.client.GetObject(ctx, e.bucket, key, minio.GetObjectOptions{})
@@ -184,20 +199,29 @@ func (e *S3Engine) AddBlock(ctx context.Context, slot uint64, root []byte, dataC
 	}
 
 	metadata := &objectMetadata{
-		objVersion:   uint32(blockData.HeaderVersion),
+		objVersion:   1,
 		headerLength: uint32(len(blockData.HeaderData)),
 		bodyVersion:  uint32(blockData.BodyVersion),
 		bodyLength:   uint32(len(blockData.BodyData)),
 	}
 
+	if blockData.PayloadVersion != 0 {
+		metadata.objVersion = 2
+		metadata.payloadVersion = uint32(blockData.PayloadVersion)
+		metadata.payloadLength = uint32(len(blockData.PayloadData))
+	}
+
 	metadataBytes := e.writeObjectMetadata(metadata)
 	metadataLength := len(metadataBytes)
 
 	// Prepare data with header and body versions and lengths
-	data := make([]byte, metadataLength+int(metadata.headerLength)+int(metadata.bodyLength))
+	data := make([]byte, metadataLength+int(metadata.headerLength)+int(metadata.bodyLength)+int(metadata.payloadLength))
 	copy(data[:metadataLength], metadataBytes)
 	copy(data[metadataLength:metadataLength+int(metadata.headerLength)], blockData.HeaderData)
-	copy(data[metadataLength+int(metadata.headerLength):], blockData.BodyData)
+	copy(data[metadataLength+int(metadata.headerLength):metadataLength+int(metadata.headerLength)+int(metadata.bodyLength)], blockData.BodyData)
+	if metadata.objVersion == 2 {
+		copy(data[metadataLength+int(metadata.headerLength)+int(metadata.bodyLength):metadataLength+int(metadata.headerLength)+int(metadata.bodyLength)+int(metadata.payloadLength)], blockData.PayloadData)
+	}
 
 	// Upload object
 	_, err = e.client.PutObject(

blockdb/types/engine.go

@@ -3,14 +3,17 @@ package types
 import "context"
 
 type BlockData struct {
-	HeaderVersion uint64
-	HeaderData    []byte
-	BodyVersion   uint64
-	BodyData      []byte
-	Body          interface{}
+	HeaderVersion  uint64
+	HeaderData     []byte
+	BodyVersion    uint64
+	BodyData       []byte
+	Body           interface{}
+	PayloadVersion uint64
+	PayloadData    []byte
+	Payload        interface{}
 }
 type BlockDbEngine interface {
 	Close() error
-	GetBlock(ctx context.Context, slot uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error)) (*BlockData, error)
+	GetBlock(ctx context.Context, slot uint64, root []byte, parseBlock func(uint64, []byte) (interface{}, error), parsePayload func(uint64, []byte) (interface{}, error)) (*BlockData, error)
 	AddBlock(ctx context.Context, slot uint64, root []byte, dataCb func() (*BlockData, error)) (bool, error)
 }

clients/consensus/chainspec.go

@@ -36,6 +36,8 @@ type ChainSpec struct {
 	ElectraForkEpoch                      *uint64           `yaml:"ELECTRA_FORK_EPOCH"   check-if-fork:"ElectraForkEpoch"`
 	Eip7594ForkVersion                    phase0.Version    `yaml:"EIP7594_FORK_VERSION" check-if-fork:"Eip7594ForkEpoch"`
 	Eip7594ForkEpoch                      *uint64           `yaml:"EIP7594_FORK_EPOCH"   check-if-fork:"Eip7594ForkEpoch"`
+	Eip7732ForkVersion                    phase0.Version    `yaml:"EIP7732_FORK_VERSION" check-if-fork:"Eip7732ForkEpoch"`
+	Eip7732ForkEpoch                      *uint64           `yaml:"EIP7732_FORK_EPOCH"`
 	SecondsPerSlot                        time.Duration     `yaml:"SECONDS_PER_SLOT"`
 	SlotsPerEpoch                         uint64            `yaml:"SLOTS_PER_EPOCH"`
 	EpochsPerHistoricalVector             uint64            `yaml:"EPOCHS_PER_HISTORICAL_VECTOR"`
@@ -72,6 +74,11 @@ type ChainSpec struct {
 	DataColumnSidecarSubnetCount *uint64 `yaml:"DATA_COLUMN_SIDECAR_SUBNET_COUNT" check-if-fork:"Eip7594ForkEpoch"`
 	CustodyRequirement           *uint64 `yaml:"CUSTODY_REQUIREMENT"              check-if-fork:"Eip7594ForkEpoch"`
 
+	// EIP7732: ePBS
+	PtcSize                uint64            `yaml:"PTC_SIZE" check-if-fork:"Eip7732ForkEpoch"`
+	MaxPayloadAttestations uint64            `yaml:"MAX_PAYLOAD_ATTESTATIONS" check-if-fork:"Eip7732ForkEpoch"`
+	DomainPtcAttester      phase0.DomainType `yaml:"DOMAIN_PTC_ATTESTER" check-if-fork:"Eip7732ForkEpoch"`
+
 	// additional dora specific specs
 	WhiskForkEpoch *uint64
 }

clients/consensus/chainstate.go

@@ -267,6 +267,14 @@ func (cs *ChainState) GetValidatorChurnLimit(validatorCount uint64) uint64 {
 	return adaptable
 }
 
+func (cs *ChainState) IsEip7732Enabled(epoch phase0.Epoch) bool {
+	if cs.specs == nil {
+		return false
+	}
+
+	return cs.specs.Eip7732ForkEpoch != nil && phase0.Epoch(*cs.specs.Eip7732ForkEpoch) <= epoch
+}
+
 func (cs *ChainState) GetBalanceChurnLimit(totalActiveBalance uint64) uint64 {
 	if cs.specs == nil {
 		return 0

clients/consensus/client.go

@@ -23,37 +23,38 @@ type ClientConfig struct {
 }
 
 type Client struct {
-	pool                    *Pool
-	clientIdx               uint16
-	endpointConfig          *ClientConfig
-	clientCtx               context.Context
-	clientCtxCancel         context.CancelFunc
-	rpcClient               *rpc.BeaconClient
-	logger                  *logrus.Entry
-	isOnline                bool
-	isSyncing               bool
-	isOptimistic            bool
-	versionStr              string
-	nodeIdentity            *rpc.NodeIdentity
-	clientType              ClientType
-	lastEvent               time.Time
-	retryCounter            uint64
-	lastError               error
-	headMutex               sync.RWMutex
-	headRoot                phase0.Root
-	headSlot                phase0.Slot
-	justifiedRoot           phase0.Root
-	justifiedEpoch          phase0.Epoch
-	finalizedRoot           phase0.Root
-	finalizedEpoch          phase0.Epoch
-	lastFinalityUpdateEpoch phase0.Epoch
-	lastMetadataUpdateEpoch phase0.Epoch
-	lastMetadataUpdateTime  time.Time
-	lastSyncUpdateEpoch     phase0.Epoch
-	peers                   []*v1.Peer
-	blockDispatcher         utils.Dispatcher[*v1.BlockEvent]
-	headDispatcher          utils.Dispatcher[*v1.HeadEvent]
-	checkpointDispatcher    utils.Dispatcher[*v1.Finality]
+	pool                       *Pool
+	clientIdx                  uint16
+	endpointConfig             *ClientConfig
+	clientCtx                  context.Context
+	clientCtxCancel            context.CancelFunc
+	rpcClient                  *rpc.BeaconClient
+	logger                     *logrus.Entry
+	isOnline                   bool
+	isSyncing                  bool
+	isOptimistic               bool
+	versionStr                 string
+	nodeIdentity               *rpc.NodeIdentity
+	clientType                 ClientType
+	lastEvent                  time.Time
+	retryCounter               uint64
+	lastError                  error
+	headMutex                  sync.RWMutex
+	headRoot                   phase0.Root
+	headSlot                   phase0.Slot
+	justifiedRoot              phase0.Root
+	justifiedEpoch             phase0.Epoch
+	finalizedRoot              phase0.Root
+	finalizedEpoch             phase0.Epoch
+	lastFinalityUpdateEpoch    phase0.Epoch
+	lastMetadataUpdateEpoch    phase0.Epoch
+	lastMetadataUpdateTime     time.Time
+	lastSyncUpdateEpoch        phase0.Epoch
+	peers                      []*v1.Peer
+	blockDispatcher            utils.Dispatcher[*v1.BlockEvent]
+	headDispatcher             utils.Dispatcher[*v1.HeadEvent]
+	checkpointDispatcher       utils.Dispatcher[*v1.Finality]
+	executionPayloadDispatcher utils.Dispatcher[*v1.ExecutionPayloadEvent]
 }
 
 func (pool *Pool) newPoolClient(clientIdx uint16, endpoint *ClientConfig) (*Client, error) {
@@ -98,6 +99,10 @@ func (client *Client) SubscribeFinalizedEvent(capacity int) *utils.Subscription[
 	return client.checkpointDispatcher.Subscribe(capacity, false)
 }
 
+func (client *Client) SubscribeExecutionPayloadEvent(capacity int, blocking bool) *utils.Subscription[*v1.ExecutionPayloadEvent] {
+	return client.executionPayloadDispatcher.Subscribe(capacity, blocking)
+}
+
 func (client *Client) GetPool() *Pool {
 	return client.pool
 }

clients/consensus/clientlogic.go

@@ -127,7 +127,11 @@ func (client *Client) runClientLogic() error {
 	}
 
 	// start event stream
-	blockStream := client.rpcClient.NewBlockStream(client.clientCtx, client.logger, rpc.StreamBlockEvent|rpc.StreamHeadEvent|rpc.StreamFinalizedEvent)
+	blockStream := client.rpcClient.NewBlockStream(
+		client.clientCtx,
+		client.logger,
+		rpc.StreamBlockEvent|rpc.StreamHeadEvent|rpc.StreamFinalizedEvent|rpc.StreamExecutionPayloadEvent,
+	)
 	defer blockStream.Close()
 
 	// process events
@@ -165,6 +169,12 @@ func (client *Client) runClientLogic() error {
 				if err != nil {
 					client.logger.Warnf("failed processing finalized event: %v", err)
 				}
+
+			case rpc.StreamExecutionPayloadEvent:
+				err := client.processExecutionPayloadEvent(evt.Data.(*v1.ExecutionPayloadEvent))
+				if err != nil {
+					client.logger.Warnf("failed processing execution payload event: %v", err)
+				}
 			}
 
 			client.logger.Tracef("event (%v) processing time: %v ms", evt.Event, time.Since(now).Milliseconds())
@@ -403,3 +413,9 @@ func (client *Client) pollClientHead() error {
 
 	return nil
 }
+
+func (client *Client) processExecutionPayloadEvent(evt *v1.ExecutionPayloadEvent) error {
+	client.executionPayloadDispatcher.Fire(evt)
+
+	return nil
+}