Merge pull request #234 from ethpandaops/pk910/eip7732-workspace

WIP: full ePBS implementation

Author: pk910

clients/consensus/chainstate.go

@@ -265,3 +265,11 @@ 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
+}

clients/consensus/client.go

@@ -22,36 +22,37 @@ 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
-	lastPeerUpdateEpoch     phase0.Epoch
-	lastSyncUpdateEpoch     phase0.Epoch
-	peers                   []*v1.Peer
-	blockDispatcher         Dispatcher[*v1.BlockEvent]
-	headDispatcher          Dispatcher[*v1.HeadEvent]
-	checkpointDispatcher    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
+	lastPeerUpdateEpoch        phase0.Epoch
+	lastSyncUpdateEpoch        phase0.Epoch
+	peers                      []*v1.Peer
+	blockDispatcher            Dispatcher[*v1.BlockEvent]
+	headDispatcher             Dispatcher[*v1.HeadEvent]
+	checkpointDispatcher       Dispatcher[*v1.Finality]
+	executionPayloadDispatcher Dispatcher[*v1.ExecutionPayloadEvent]
 }
 
 func (pool *Pool) newPoolClient(clientIdx uint16, endpoint *ClientConfig) (*Client, error) {
@@ -96,6 +97,10 @@ func (client *Client) SubscribeFinalizedEvent(capacity int) *Subscription[*v1.Fi
 	return client.checkpointDispatcher.Subscribe(capacity, false)
 }
 
+func (client *Client) SubscribeExecutionPayloadEvent(capacity int, blocking bool) *Subscription[*v1.ExecutionPayloadEvent] {
+	return client.executionPayloadDispatcher.Subscribe(capacity, blocking)
+}
+
 func (client *Client) GetPool() *Pool {
 	return client.pool
 }

clients/consensus/clientlogic.go

@@ -133,7 +133,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
@@ -171,6 +175,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())
@@ -392,3 +402,9 @@ func (client *Client) pollClientHead() error {
 
 	return nil
 }
+
+func (client *Client) processExecutionPayloadEvent(evt *v1.ExecutionPayloadEvent) error {
+	client.executionPayloadDispatcher.Fire(evt)
+
+	return nil
+}

clients/consensus/rpc/beaconapi.go

@@ -19,6 +19,7 @@ import (
 	"github.com/attestantio/go-eth2-client/spec"
 	"github.com/attestantio/go-eth2-client/spec/capella"
 	"github.com/attestantio/go-eth2-client/spec/deneb"
+	"github.com/attestantio/go-eth2-client/spec/eip7732"
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	"github.com/rs/zerolog"
 	"github.com/sirupsen/logrus"
@@ -406,6 +407,22 @@ func (bc *BeaconClient) GetBlockBodyByBlockroot(ctx context.Context, blockroot p
 	return result.Data, nil
 }
 
+func (bc *BeaconClient) GetExecutionPayloadByBlockroot(ctx context.Context, blockroot phase0.Root) (*eip7732.SignedExecutionPayloadEnvelope, error) {
+	provider, isProvider := bc.clientSvc.(eth2client.ExecutionPayloadProvider)
+	if !isProvider {
+		return nil, fmt.Errorf("get execution payload not supported")
+	}
+
+	result, err := provider.SignedExecutionPayloadEnvelope(ctx, &api.SignedExecutionPayloadEnvelopeOpts{
+		Block: fmt.Sprintf("0x%x", blockroot),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return result.Data, nil
+}
+
 func (bc *BeaconClient) GetState(ctx context.Context, stateRef string) (*spec.VersionedBeaconState, error) {
 	provider, isProvider := bc.clientSvc.(eth2client.BeaconStateProvider)
 	if !isProvider {

clients/consensus/rpc/beaconstream.go

@@ -17,9 +17,10 @@ import (
 )
 
 const (
-	StreamBlockEvent     uint16 = 0x01
-	StreamHeadEvent      uint16 = 0x02
-	StreamFinalizedEvent uint16 = 0x04
+	StreamBlockEvent            uint16 = 0x01
+	StreamHeadEvent             uint16 = 0x02
+	StreamFinalizedEvent        uint16 = 0x04
+	StreamExecutionPayloadEvent uint16 = 0x08
 )
 
 type BeaconStreamEvent struct {
@@ -87,6 +88,8 @@ func (bs *BeaconStream) startStream() {
 					bs.processHeadEvent(evt)
 				case "finalized_checkpoint":
 					bs.processFinalizedEvent(evt)
+				case "execution_payload":
+					bs.processExecutionPayloadEvent(evt)
 				}
 			case <-stream.Ready:
 				bs.ReadyChan <- &BeaconStreamStatus{
@@ -148,6 +151,16 @@ func (bs *BeaconStream) subscribeStream(endpoint string, events uint16) *eventst
 		topicsCount++
 	}
 
+	if events&StreamExecutionPayloadEvent > 0 {
+		if topicsCount > 0 {
+			fmt.Fprintf(&topics, ",")
+		}
+
+		fmt.Fprintf(&topics, "execution_payload")
+
+		topicsCount++
+	}
+
 	if topicsCount == 0 {
 		return nil
 	}
@@ -225,6 +238,21 @@ func (bs *BeaconStream) processFinalizedEvent(evt eventsource.Event) {
 	}
 }
 
+func (bs *BeaconStream) processExecutionPayloadEvent(evt eventsource.Event) {
+	var parsed v1.ExecutionPayloadEvent
+
+	err := json.Unmarshal([]byte(evt.Data()), &parsed)
+	if err != nil {
+		bs.logger.Warnf("beacon block stream failed to decode execution_payload event: %v", err)
+		return
+	}
+
+	bs.EventChan <- &BeaconStreamEvent{
+		Event: StreamExecutionPayloadEvent,
+		Data:  &parsed,
+	}
+}
+
 func getRedactedURL(requrl string) string {
 	var logurl string
 

db/epochs.go

@@ -11,8 +11,8 @@ func InsertEpoch(epoch *dbtypes.Epoch, tx *sqlx.Tx) error {
 			INSERT INTO epochs (
 				epoch, validator_count, validator_balance, eligible, voted_target, voted_head, voted_total, block_count, orphaned_count,
 				attestation_count, deposit_count, exit_count, withdraw_count, withdraw_amount, attester_slashing_count, 
-				proposer_slashing_count, bls_change_count, eth_transaction_count, sync_participation
-			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)
+				proposer_slashing_count, bls_change_count, eth_transaction_count, sync_participation, payload_count
+			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)
 			ON CONFLICT (epoch) DO UPDATE SET
 				validator_count = excluded.validator_count,
 				validator_balance = excluded.validator_balance,
@@ -31,17 +31,18 @@ func InsertEpoch(epoch *dbtypes.Epoch, tx *sqlx.Tx) error {
 				proposer_slashing_count = excluded.proposer_slashing_count, 
 				bls_change_count = excluded.bls_change_count, 
 				eth_transaction_count = excluded.eth_transaction_count, 
-				sync_participation = excluded.sync_participation`,
+				sync_participation = excluded.sync_participation,
+				payload_count = excluded.payload_count`,
 		dbtypes.DBEngineSqlite: `
 			INSERT OR REPLACE INTO epochs (
 				epoch, validator_count, validator_balance, eligible, voted_target, voted_head, voted_total, block_count, orphaned_count,
 				attestation_count, deposit_count, exit_count, withdraw_count, withdraw_amount, attester_slashing_count, 
-				proposer_slashing_count, bls_change_count, eth_transaction_count, sync_participation
-			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)`,
+				proposer_slashing_count, bls_change_count, eth_transaction_count, sync_participation, payload_count
+			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)`,
 	}),
 		epoch.Epoch, epoch.ValidatorCount, epoch.ValidatorBalance, epoch.Eligible, epoch.VotedTarget, epoch.VotedHead, epoch.VotedTotal, epoch.BlockCount, epoch.OrphanedCount,
 		epoch.AttestationCount, epoch.DepositCount, epoch.ExitCount, epoch.WithdrawCount, epoch.WithdrawAmount, epoch.AttesterSlashingCount, epoch.ProposerSlashingCount,
-		epoch.BLSChangeCount, epoch.EthTransactionCount, epoch.SyncParticipation)
+		epoch.BLSChangeCount, epoch.EthTransactionCount, epoch.SyncParticipation, epoch.PayloadCount)
 	if err != nil {
 		return err
 	}
@@ -63,7 +64,7 @@ func GetEpochs(firstEpoch uint64, limit uint32) []*dbtypes.Epoch {
 	SELECT
 		epoch, validator_count, validator_balance, eligible, voted_target, voted_head, voted_total, block_count, orphaned_count,
 		attestation_count, deposit_count, exit_count, withdraw_count, withdraw_amount, attester_slashing_count,
-		proposer_slashing_count, bls_change_count, eth_transaction_count, sync_participation
+		proposer_slashing_count, bls_change_count, eth_transaction_count, sync_participation, payload_count
 	FROM epochs
 	WHERE epoch <= $1
 	ORDER BY epoch DESC

db/orphaned_blocks.go

@@ -9,15 +9,15 @@ func InsertOrphanedBlock(block *dbtypes.OrphanedBlock, tx *sqlx.Tx) error {
 	_, err := tx.Exec(EngineQuery(map[dbtypes.DBEngineType]string{
 		dbtypes.DBEnginePgsql: `
 			INSERT INTO orphaned_blocks (
-				root, header_ver, header_ssz, block_ver, block_ssz
-			) VALUES ($1, $2, $3, $4, $5)
+				root, header_ver, header_ssz, block_ver, block_ssz, payload_ver, payload_ssz
+			) VALUES ($1, $2, $3, $4, $5, $6, $7)
 			ON CONFLICT (root) DO NOTHING`,
 		dbtypes.DBEngineSqlite: `
 			INSERT OR IGNORE INTO orphaned_blocks (
-				root, header_ver, header_ssz, block_ver, block_ssz
-			) VALUES ($1, $2, $3, $4, $5)`,
+				root, header_ver, header_ssz, block_ver, block_ssz, payload_ver, payload_ssz
+			) VALUES ($1, $2, $3, $4, $5, $6, $7)`,
 	}),
-		block.Root, block.HeaderVer, block.HeaderSSZ, block.BlockVer, block.BlockSSZ)
+		block.Root, block.HeaderVer, block.HeaderSSZ, block.BlockVer, block.BlockSSZ, block.PayloadVer, block.PayloadSSZ)
 	if err != nil {
 		return err
 	}
@@ -27,7 +27,7 @@ func InsertOrphanedBlock(block *dbtypes.OrphanedBlock, tx *sqlx.Tx) error {
 func GetOrphanedBlock(root []byte) *dbtypes.OrphanedBlock {
 	block := dbtypes.OrphanedBlock{}
 	err := ReaderDb.Get(&block, `
-	SELECT root, header_ver, header_ssz, block_ver, block_ssz
+	SELECT root, header_ver, header_ssz, block_ver, block_ssz, payload_ver, payload_ssz
 	FROM orphaned_blocks
 	WHERE root = $1
 	`, root)

db/schema/pgsql/20250208225212_epbs-payload.sql

@@ -0,0 +1,29 @@
+-- +goose Up
+-- +goose StatementBegin
+
+ALTER TABLE public."unfinalized_blocks" ADD
+ "payload_ver" int NOT NULL DEFAULT 0,
+ "payload_ssz" bytea NULL;
+
+ALTER TABLE public."orphaned_blocks" ADD
+ "payload_ver" int NOT NULL DEFAULT 0,
+ "payload_ssz" bytea NULL;
+
+ALTER TABLE public."slots" ADD
+ "has_payload" boolean NOT NULL DEFAULT false;
+
+CREATE INDEX IF NOT EXISTS "slots_has_payload_idx"
+    ON public."slots" 
+    ("has_payload" ASC NULLS LAST);
+
+ALTER TABLE public."epochs" ADD
+ "payload_count" int NOT NULL DEFAULT 0;
+
+ALTER TABLE public."unfinalized_epochs" ADD
+ "payload_count" int NOT NULL DEFAULT 0;
+
+-- +goose StatementEnd
+-- +goose Down
+-- +goose StatementBegin
+SELECT 'NOT SUPPORTED';
+-- +goose StatementEnd

db/schema/sqlite/20250208225212_epbs-payload.sql

@@ -0,0 +1,22 @@
+-- +goose Up
+-- +goose StatementBegin
+
+ALTER TABLE "unfinalized_blocks" ADD "payload_ver" int NOT NULL DEFAULT 0;
+ALTER TABLE "unfinalized_blocks" ADD "payload_ssz" BLOB NULL;
+
+ALTER TABLE "orphaned_blocks" ADD "payload_ver" int NOT NULL DEFAULT 0;
+ALTER TABLE "orphaned_blocks" ADD "payload_ssz" BLOB NULL;
+
+ALTER TABLE "slots" ADD "has_payload" boolean NOT NULL DEFAULT false;
+
+CREATE INDEX IF NOT EXISTS "slots_has_payload_idx" ON "slots" ("has_payload" ASC);
+
+ALTER TABLE "epochs" ADD "payload_count" int NOT NULL DEFAULT 0;
+
+ALTER TABLE "unfinalized_epochs" ADD "payload_count" int NOT NULL DEFAULT 0;
+
+-- +goose StatementEnd
+-- +goose Down
+-- +goose StatementBegin
+SELECT 'NOT SUPPORTED';
+-- +goose StatementEnd