Skip to content
This repository was archived by the owner on Oct 25, 2024. It is now read-only.

Add SSE subscription to builder #53

Merged
avalonche merged 6 commits into from
Apr 10, 2023
Merged

Add SSE subscription to builder #53

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
27c13dd
Add SSE subscription to builder
avalonche Mar 23, 2023
0bd6488
withdrawals marshalling
avalonche Mar 24, 2023
18c472d
add stop channel
avalonche Mar 25, 2023
ab0434c
pr comments
avalonche Apr 3, 2023
f51d92f
Add handling multiple beacon clients (#57)
Ruteri Apr 10, 2023
fd689ce
linting
avalonche Apr 10, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 88 additions & 2 deletions builder/beacon_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,22 @@ import (
"sync"
"time"

"github.com/attestantio/go-eth2-client/spec/capella"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log"
"github.com/r3labs/sse"
)

type IBeaconClient interface {
isValidator(pubkey PubkeyHex) bool
getProposerForNextSlot(requestedSlot uint64) (PubkeyHex, error)
SubscribeToPayloadAttributesEvents(payloadAttrC chan types.BuilderPayloadAttributes)
Start() error
Stop()
}

type testBeaconClient struct {
validator *ValidatorPrivateData
slot uint64
Expand All @@ -30,10 +41,29 @@ func (b *testBeaconClient) isValidator(pubkey PubkeyHex) bool {
func (b *testBeaconClient) getProposerForNextSlot(requestedSlot uint64) (PubkeyHex, error) {
return PubkeyHex(hexutil.Encode(b.validator.Pk)), nil
}
func (b *testBeaconClient) Start() error {
return nil

func (b *testBeaconClient) SubscribeToPayloadAttributesEvents(payloadAttrC chan types.BuilderPayloadAttributes) {
}

func (b *testBeaconClient) Start() error { return nil }

type NilBeaconClient struct{}

func (b *NilBeaconClient) isValidator(pubkey PubkeyHex) bool {
return false
}

func (b *NilBeaconClient) getProposerForNextSlot(requestedSlot uint64) (PubkeyHex, error) {
return PubkeyHex(""), nil
}

func (b *NilBeaconClient) SubscribeToPayloadAttributesEvents(payloadAttrC chan types.BuilderPayloadAttributes) {
}

func (b *NilBeaconClient) Start() error { return nil }

func (b *NilBeaconClient) Stop() {}

type BeaconClient struct {
endpoint string
slotsInEpoch uint64
Expand Down Expand Up @@ -256,3 +286,59 @@ func fetchBeacon(url string, dst any) error {
log.Info("fetched", "url", url, "res", dst)
return nil
}

// SubscribeToPayloadAttributesEvents subscribes to payload attributes events to validate fields such as prevrandao and withdrawals
func (b *BeaconClient) SubscribeToPayloadAttributesEvents(payloadAttrC chan types.BuilderPayloadAttributes) {
payloadAttributesResp := &struct {
Copy link
Collaborator

@Ruteri Ruteri Mar 24, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This got to be put somewhere as a regular structure :)

Copy link
Contributor Author

@avalonche avalonche Mar 25, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will add in attestant types!

Version string `json:"version"`
Data struct {
ProposalSlot uint64 `json:"proposal_slot,string"`
ParentBlockHash common.Hash `json:"parent_block_hash"`
PayloadAttributes struct {
Timestamp uint64 `json:"timestamp,string"`
PrevRandao common.Hash `json:"prev_randao"`
SuggestedFeeRecipient common.Address `json:"suggested_fee_recipient"`
Withdrawals []*capella.Withdrawal `json:"withdrawals"`
} `json:"payload_attributes"`
} `json:"data"`
}{}

eventsURL := fmt.Sprintf("%s/eth/v1/events?topics=payload_attributes", b.endpoint)
log.Info("subscribing to payload_attributes events")

for {
client := sse.NewClient(eventsURL)
err := client.SubscribeRaw(func(msg *sse.Event) {
Copy link
Collaborator

@Ruteri Ruteri Mar 24, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it be better if the callback was a standalone function?

err := json.Unmarshal(msg.Data, &payloadAttributesResp)
if err != nil {
log.Error("could not unmarshal payload_attributes event", "err", err)
} else {
// convert capella.Withdrawal to types.Withdrawal
withdrawals := make([]*types.Withdrawal, len(payloadAttributesResp.Data.PayloadAttributes.Withdrawals))
for i, w := range payloadAttributesResp.Data.PayloadAttributes.Withdrawals {
withdrawals[i] = &types.Withdrawal{
Index: uint64(w.Index),
Validator: uint64(w.ValidatorIndex),
Address: common.Address(w.Address),
Amount: uint64(w.Amount),
}
}

data := types.BuilderPayloadAttributes{
Slot: payloadAttributesResp.Data.ProposalSlot,
HeadHash: payloadAttributesResp.Data.ParentBlockHash,
Timestamp: hexutil.Uint64(payloadAttributesResp.Data.PayloadAttributes.Timestamp),
Random: payloadAttributesResp.Data.PayloadAttributes.PrevRandao,
SuggestedFeeRecipient: payloadAttributesResp.Data.PayloadAttributes.SuggestedFeeRecipient,
Withdrawals: withdrawals,
}
payloadAttrC <- data
}
})
if err != nil {
log.Error("failed to subscribe to payload_attributes events", "err", err)
time.Sleep(1 * time.Second)
}
log.Warn("beaconclient SubscribeRaw ended, reconnecting")
}
}
28 changes: 20 additions & 8 deletions builder/builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,6 @@ type ValidatorData struct {
GasLimit uint64
}

type IBeaconClient interface {
isValidator(pubkey PubkeyHex) bool
getProposerForNextSlot(requestedSlot uint64) (PubkeyHex, error)
Start() error
Stop()
}

type IRelay interface {
SubmitBlock(msg *boostTypes.BuilderSubmitBlockRequest, vd ValidatorData) error
SubmitBlockCapella(msg *capellaapi.SubmitBlockRequest, vd ValidatorData) error
Expand All @@ -64,6 +57,7 @@ type Builder struct {
eth IEthereumService
dryRun bool
validator *blockvalidation.BlockValidationAPI
beaconClient IBeaconClient
builderSecretKey *bls.SecretKey
builderPublicKey boostTypes.PublicKey
builderSigningDomain boostTypes.Domain
Expand All @@ -75,9 +69,11 @@ type Builder struct {
slotAttrs []types.BuilderPayloadAttributes
slotCtx context.Context
slotCtxCancel context.CancelFunc

stop chan struct{}
}

func NewBuilder(sk *bls.SecretKey, ds flashbotsextra.IDatabaseService, relay IRelay, builderSigningDomain boostTypes.Domain, eth IEthereumService, dryRun bool, validator *blockvalidation.BlockValidationAPI) *Builder {
func NewBuilder(sk *bls.SecretKey, ds flashbotsextra.IDatabaseService, relay IRelay, builderSigningDomain boostTypes.Domain, eth IEthereumService, dryRun bool, validator *blockvalidation.BlockValidationAPI, beaconClient IBeaconClient) *Builder {
pkBytes := bls.PublicKeyFromSecretKey(sk).Compress()
pk := boostTypes.PublicKey{}
pk.FromSlice(pkBytes)
Expand All @@ -89,6 +85,7 @@ func NewBuilder(sk *bls.SecretKey, ds flashbotsextra.IDatabaseService, relay IRe
eth: eth,
dryRun: dryRun,
validator: validator,
beaconClient: beaconClient,
builderSecretKey: sk,
builderPublicKey: pk,
builderSigningDomain: builderSigningDomain,
Expand All @@ -101,10 +98,25 @@ func NewBuilder(sk *bls.SecretKey, ds flashbotsextra.IDatabaseService, relay IRe
}

func (b *Builder) Start() error {
// Start regular payload attributes updates
go func() {
c := make(chan types.BuilderPayloadAttributes)
go b.beaconClient.SubscribeToPayloadAttributesEvents(c)
for {
select {
case <-b.stop:
break
default:
payloadAttributes := <-c
b.OnPayloadAttribute(&payloadAttributes)
}
}
}()
return b.relay.Start()
}

func (b *Builder) Stop() error {
close(b.stop)
return nil
}

Expand Down
2 changes: 1 addition & 1 deletion builder/builder_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ func TestOnPayloadAttributes(t *testing.T) {

testEthService := &testEthereumService{synced: true, testExecutableData: testExecutableData, testBlock: testBlock, testBlockValue: big.NewInt(10)}

builder := NewBuilder(sk, flashbotsextra.NilDbService{}, &testRelay, bDomain, testEthService, false, nil)
builder := NewBuilder(sk, flashbotsextra.NilDbService{}, &testRelay, bDomain, testEthService, false, nil, &testBeacon)
builder.Start()
defer builder.Stop()

Expand Down
2 changes: 1 addition & 1 deletion builder/local_relay_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ func newTestBackend(t *testing.T, forkchoiceData *engine.ExecutableData, block *
beaconClient := &testBeaconClient{validator: validator}
localRelay := NewLocalRelay(sk, beaconClient, bDomain, cDomain, ForkData{}, true)
ethService := &testEthereumService{synced: true, testExecutableData: forkchoiceData, testBlock: block, testBlockValue: blockValue}
backend := NewBuilder(sk, flashbotsextra.NilDbService{}, localRelay, bDomain, ethService, false, nil)
backend := NewBuilder(sk, flashbotsextra.NilDbService{}, localRelay, bDomain, ethService, false, nil, beaconClient)
// service := NewService("127.0.0.1:31545", backend)

backend.limiter = rate.NewLimiter(rate.Inf, 0)
Expand Down
9 changes: 7 additions & 2 deletions builder/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,12 @@ func Register(stack *node.Node, backend *eth.Ethereum, cfg *Config) error {
copy(bellatrixForkVersion[:], bellatrixForkVersionBytes[:4])
proposerSigningDomain := boostTypes.ComputeDomain(boostTypes.DomainTypeBeaconProposer, bellatrixForkVersion, genesisValidatorsRoot)

beaconClient := NewBeaconClient(cfg.BeaconEndpoint, cfg.SlotsInEpoch, cfg.SecondsInSlot)
var beaconClient IBeaconClient
if cfg.BeaconEndpoint != "" {
Copy link
Collaborator

@Ruteri Ruteri Mar 24, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does it make sense to run the builder service without a beacon configured?

Copy link
Contributor Author

@avalonche avalonche Mar 24, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that would break backwards compatibility. This allows the optionality to use the our old method of prysm block building during testing by not subscribing to the SSE endpoint.

Copy link
Collaborator

@Ruteri Ruteri Apr 3, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why though?
The old code seems to always construct a beacon client

Copy link
Contributor Author

@avalonche avalonche Apr 3, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the old code would construct the beacon client with an empty url and only fail if the local relay is used.

beaconClient = NewBeaconClient(cfg.BeaconEndpoint, cfg.SlotsInEpoch, cfg.SecondsInSlot)
} else {
beaconClient = &NilBeaconClient{}
}

var localRelay *LocalRelay
if cfg.EnableLocalRelay {
Expand Down Expand Up @@ -196,7 +201,7 @@ func Register(stack *node.Node, backend *eth.Ethereum, cfg *Config) error {
return errors.New("incorrect builder API secret key provided")
}

builderBackend := NewBuilder(builderSk, ds, relay, builderSigningDomain, ethereumService, cfg.DryRun, validator)
builderBackend := NewBuilder(builderSk, ds, relay, builderSigningDomain, ethereumService, cfg.DryRun, validator, beaconClient)
builderService := NewService(cfg.ListenAddr, localRelay, builderBackend)

stack.RegisterAPIs([]rpc.API{
Expand Down
2 changes: 1 addition & 1 deletion core/txpool/txpool_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2447,7 +2447,7 @@ func TestSlotCount(t *testing.T) {
func TestBundleCancellations(t *testing.T) {
// Create the pool to test the status retrievals with
statedb, _ := state.New(common.Hash{}, state.NewDatabase(rawdb.NewMemoryDatabase()), nil)
blockchain := &testBlockChain{100, statedb, new(event.Feed)}
blockchain := newTestBlockChain(100, statedb, new(event.Feed))

pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain)
fetcher := &mockFetcher{make(map[int64]error), make(map[int64][]types.LatestUuidBundle)}
Expand Down
3 changes: 3 additions & 0 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,8 @@ require (
gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce
)

require gopkg.in/cenkalti/backoff.v1 v1.1.0 // indirect

require (
github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1 // indirect
github.com/Azure/azure-sdk-for-go/sdk/internal v0.8.3 // indirect
Expand Down Expand Up @@ -125,6 +127,7 @@ require (
github.com/prometheus/common v0.39.0 // indirect
github.com/prometheus/procfs v0.9.0 // indirect
github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 // indirect
github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc
github.com/rogpeppe/go-internal v1.9.0 // indirect
github.com/russross/blackfriday/v2 v2.1.0 // indirect
github.com/sirupsen/logrus v1.9.0 // indirect
Expand Down
3 changes: 3 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -595,6 +595,8 @@ github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 h1:0tVE4
github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7/go.mod h1:wmuf/mdK4VMD+jA9ThwcUKjg3a2XWM9cVfFYjDyY4j4=
github.com/prysmaticlabs/gohashtree v0.0.1-alpha.0.20220714111606-acbb2962fb48 h1:cSo6/vk8YpvkLbk9v3FO97cakNmUoxwi2KMP8hd5WIw=
github.com/prysmaticlabs/gohashtree v0.0.1-alpha.0.20220714111606-acbb2962fb48/go.mod h1:4pWaT30XoEx1j8KNJf3TV+E3mQkaufn7mf+jRNb/Fuk=
github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc h1:zAsgcP8MhzAbhMnB1QQ2O7ZhWYVGYSR2iVcjzQuPV+o=
github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc/go.mod h1:S8xSOnV3CgpNrWd0GQ/OoQfMtlg2uPRSuTzcSGrzwK8=
github.com/r3labs/sse/v2 v2.7.4/go.mod h1:hUrYMKfu9WquG9MyI0r6TKiNH+6Sw/QPKm2YbNbU5g8=
github.com/retailnext/hllpp v1.0.1-0.20180308014038-101a6d2f8b52/go.mod h1:RDpi1RftBQPUCDRw6SmxeaREsAaRKnOclghuzp/WRzc=
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
Expand Down Expand Up @@ -1065,6 +1067,7 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
gopkg.in/cenkalti/backoff.v1 v1.1.0 h1:Arh75ttbsvlpVA7WtVpH4u9h6Zl46xuptxqLxPiSo4Y=
gopkg.in/cenkalti/backoff.v1 v1.1.0/go.mod h1:J6Vskwqd+OMVJl8C33mmtxTBs2gyzfv7UDAkHu8BrjI=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
Expand Down