Skip to content

feat(deneb): proposer rpc to handle builder flow #12554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
terencechain merged 6 commits into from
Jun 26, 2023
Merged

feat(deneb): proposer rpc to handle builder flow #12554

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d5dfbc6
feat(deneb): proposer rpc to handle builder flow
terencechain Jun 20, 2023
2c8530e
fix: refactor blobs
terencechain Jun 21, 2023
d1284cc
test: get block
terencechain Jun 21, 2023
db935ce
test: more tests
terencechain Jun 23, 2023
4a0b960
test: fix unblinder
terencechain Jun 23, 2023
f5f9478
chrone: renames
terencechain Jun 23, 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
57 changes: 40 additions & 17 deletions beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
return nil, status.Errorf(codes.Internal, "Could not get local payload: %v", err)
}

builderPayload, err := vs.getBuilderPayload(ctx, sBlk.Block().Slot(), sBlk.Block().ProposerIndex())
builderPayload, blindBlobsBundle, err := vs.getBuilderPayload(ctx, sBlk.Block().Slot(), sBlk.Block().ProposerIndex())
if err != nil {
builderGetPayloadMissCount.Inc()
log.WithError(err).Error("Could not get builder payload")
Expand All @@ -158,7 +158,7 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
return nil, status.Errorf(codes.Internal, "Could not set execution data: %v", err)
}

if err := setKzgCommitments(sBlk, blobsBundle); err != nil {
if err := setKzgCommitments(sBlk, blobsBundle, blindBlobsBundle); err != nil {
return nil, status.Errorf(codes.Internal, "Could not set kzg commitment: %v", err)
}

Expand All @@ -181,7 +181,18 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
return nil, status.Errorf(codes.Internal, "Could not convert block to proto: %v", err)
}
if slots.ToEpoch(req.Slot) >= params.BeaconConfig().DenebForkEpoch {
// TODO: Handle blind case
if sBlk.IsBlinded() {
scs, err := blindBlobsBundleToSidecars(blindBlobsBundle, sBlk.Block())
if err != nil {
return nil, status.Errorf(codes.Internal, "Could not convert blind blobs bundle to sidecar: %v", err)
}
blockAndBlobs := &ethpb.BlindedBeaconBlockAndBlobsDeneb{
Block: pb.(*ethpb.BlindedBeaconBlockDeneb),
Blobs: scs,
}
return &ethpb.GenericBeaconBlock{Block: &ethpb.GenericBeaconBlock_BlindedDeneb{BlindedDeneb: blockAndBlobs}}, nil
}

scs, err := blobsBundleToSidecars(blobsBundle, sBlk.Block())
if err != nil {
return nil, status.Errorf(codes.Internal, "Could not convert blobs bundle to sidecar: %v", err)
Expand Down Expand Up @@ -222,11 +233,17 @@ func (vs *Server) ProposeBeaconBlock(ctx context.Context, req *ethpb.GenericSign
return nil, status.Errorf(codes.InvalidArgument, "%s: %v", CouldNotDecodeBlock, err)
}

unblinder, err := newUnblinder(blk, vs.BlockBuilder)
var signedBlindBlobs []*ethpb.SignedBlindedBlobSidecar
if blk.Version() >= version.Deneb && blk.IsBlinded() {
signedBlindBlobs = req.GetBlindedDeneb().Blobs
}

unblinder, err := newUnblinder(blk, signedBlindBlobs, vs.BlockBuilder)
if err != nil {
return nil, errors.Wrap(err, "could not create unblinder")
}
blk, err = unblinder.unblindBuilderBlock(ctx)
wasBlinded := unblinder.b.IsBlinded()
blk, blobSidecar, err := unblinder.unblindBuilderBlock(ctx)
if err != nil {
return nil, errors.Wrap(err, "could not unblind builder block")
}
Expand All @@ -240,23 +257,29 @@ func (vs *Server) ProposeBeaconBlock(ctx context.Context, req *ethpb.GenericSign
return nil, fmt.Errorf("could not broadcast block: %v", err)
}

var scs []*ethpb.SignedBlobSidecar
if blk.Version() >= version.Deneb {
b, ok := req.GetBlock().(*ethpb.GenericSignedBeaconBlock_Deneb)
if !ok {
return nil, status.Error(codes.Internal, "Could not cast block to Deneb")
}
if len(b.Deneb.Blobs) > fieldparams.MaxBlobsPerBlock {
return nil, status.Errorf(codes.InvalidArgument, "Too many blobs in block: %d", len(b.Deneb.Blobs))
if wasBlinded {

@james-prysm james-prysm Jun 21, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I feel like there is a code smell around this, a lot of different arrays used, can't put my finger on how to clean it up just yet but some parts can be broken into its own function or cleaned up some way just for readability.

@terencechain terencechain Jun 21, 2023

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

slightly better: 2c8530e

scs = blobSidecar
} else {
b, ok := req.GetBlock().(*ethpb.GenericSignedBeaconBlock_Deneb)
if !ok {
return nil, status.Error(codes.Internal, "Could not cast block to Deneb")
}
if len(b.Deneb.Blobs) > fieldparams.MaxBlobsPerBlock {
return nil, status.Errorf(codes.InvalidArgument, "Too many blobs in block: %d", len(b.Deneb.Blobs))
}
scs = b.Deneb.Blobs
}
scs := make([]*ethpb.BlobSidecar, len(b.Deneb.Blobs))
for i, blob := range b.Deneb.Blobs {
if err := vs.P2P.BroadcastBlob(ctx, blob.Message.Index, blob); err != nil {
log.WithError(err).Errorf("Could not broadcast blob index %d / %d", i, len(b.Deneb.Blobs))
sidecar := make([]*ethpb.BlobSidecar, len(scs))

@james-prysm james-prysm Jun 23, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

sidecar -> sidecars?

or maybe savableSidecars?

@terencechain terencechain Jun 23, 2023

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

f5f9478

for i, sc := range scs {
if err := vs.P2P.BroadcastBlob(ctx, sc.Message.Index, sc); err != nil {
log.WithError(err).Errorf("Could not broadcast blob index %d / %d", i, len(scs))
}
scs[i] = blob.Message
sidecar[i] = sc.Message
}
if len(scs) > 0 {
if err := vs.BeaconDB.SaveBlobSidecar(ctx, scs); err != nil {
if err := vs.BeaconDB.SaveBlobSidecar(ctx, sidecar); err != nil {
return nil, err
}
}
Expand Down
52 changes: 31 additions & 21 deletions beacon-chain/rpc/prysm/v1alpha1/validator/proposer_bellatrix.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"github.com/prysmaticlabs/prysm/v4/encoding/ssz"
"github.com/prysmaticlabs/prysm/v4/monitoring/tracing"
"github.com/prysmaticlabs/prysm/v4/network/forks"
enginev1 "github.com/prysmaticlabs/prysm/v4/proto/engine/v1"
"github.com/prysmaticlabs/prysm/v4/runtime/version"
"github.com/prysmaticlabs/prysm/v4/time/slots"
"github.com/sirupsen/logrus"
Expand Down Expand Up @@ -123,89 +124,98 @@ func setExecutionData(ctx context.Context, blk interfaces.SignedBeaconBlock, loc

// This function retrieves the payload header given the slot number and the validator index.
// It's a no-op if the latest head block is not versioned bellatrix.
func (vs *Server) getPayloadHeaderFromBuilder(ctx context.Context, slot primitives.Slot, idx primitives.ValidatorIndex) (interfaces.ExecutionData, error) {
func (vs *Server) getPayloadHeaderFromBuilder(ctx context.Context, slot primitives.Slot, idx primitives.ValidatorIndex) (interfaces.ExecutionData, *enginev1.BlindedBlobsBundle, error) {
ctx, span := trace.StartSpan(ctx, "ProposerServer.getPayloadHeaderFromBuilder")
defer span.End()

if slots.ToEpoch(slot) < params.BeaconConfig().BellatrixForkEpoch {
return nil, errors.New("can't get payload header from builder before bellatrix epoch")
return nil, nil, errors.New("can't get payload header from builder before bellatrix epoch")
}

b, err := vs.HeadFetcher.HeadBlock(ctx)
if err != nil {
return nil, err
return nil, nil, err
}

h, err := b.Block().Body().Execution()
if err != nil {
return nil, errors.Wrap(err, "failed to get execution header")
return nil, nil, errors.Wrap(err, "failed to get execution header")
}
pk, err := vs.HeadFetcher.HeadValidatorIndexToPublicKey(ctx, idx)
if err != nil {
return nil, err
return nil, nil, err
}

ctx, cancel := context.WithTimeout(ctx, blockBuilderTimeout)
defer cancel()

signedBid, err := vs.BlockBuilder.GetHeader(ctx, slot, bytesutil.ToBytes32(h.BlockHash()), pk)
if err != nil {
return nil, err
return nil, nil, err
}
if signedBid.IsNil() {
return nil, errors.New("builder returned nil bid")
return nil, nil, errors.New("builder returned nil bid")
}
fork, err := forks.Fork(slots.ToEpoch(slot))
if err != nil {
return nil, errors.Wrap(err, "unable to get fork information")
return nil, nil, errors.Wrap(err, "unable to get fork information")
}
forkName, ok := params.BeaconConfig().ForkVersionNames[bytesutil.ToBytes4(fork.CurrentVersion)]
if !ok {
return nil, errors.New("unable to find current fork in schedule")
return nil, nil, errors.New("unable to find current fork in schedule")
}
if !strings.EqualFold(version.String(signedBid.Version()), forkName) {
return nil, fmt.Errorf("builder bid response version: %d is different from head block version: %d for epoch %d", signedBid.Version(), b.Version(), slots.ToEpoch(slot))
return nil, nil, fmt.Errorf("builder bid response version: %d is different from head block version: %d for epoch %d", signedBid.Version(), b.Version(), slots.ToEpoch(slot))
}

bid, err := signedBid.Message()
if err != nil {
return nil, errors.Wrap(err, "could not get bid")
return nil, nil, errors.Wrap(err, "could not get bid")
}
if bid.IsNil() {
return nil, errors.New("builder returned nil bid")
return nil, nil, errors.New("builder returned nil bid")
}

v := bytesutil.LittleEndianBytesToBigInt(bid.Value())
if v.String() == "0" {
return nil, errors.New("builder returned header with 0 bid amount")
return nil, nil, errors.New("builder returned header with 0 bid amount")
}

header, err := bid.Header()
if err != nil {
return nil, errors.Wrap(err, "could not get bid header")
return nil, nil, errors.Wrap(err, "could not get bid header")
}
txRoot, err := header.TransactionsRoot()
if err != nil {
return nil, errors.Wrap(err, "could not get transaction root")
return nil, nil, errors.Wrap(err, "could not get transaction root")
}
if bytesutil.ToBytes32(txRoot) == emptyTransactionsRoot {
return nil, errors.New("builder returned header with an empty tx root")
return nil, nil, errors.New("builder returned header with an empty tx root")
}

if !bytes.Equal(header.ParentHash(), h.BlockHash()) {
return nil, fmt.Errorf("incorrect parent hash %#x != %#x", header.ParentHash(), h.BlockHash())
return nil, nil, fmt.Errorf("incorrect parent hash %#x != %#x", header.ParentHash(), h.BlockHash())
}

t, err := slots.ToTime(uint64(vs.TimeFetcher.GenesisTime().Unix()), slot)
if err != nil {
return nil, err
return nil, nil, err
}
if header.Timestamp() != uint64(t.Unix()) {
return nil, fmt.Errorf("incorrect timestamp %d != %d", header.Timestamp(), uint64(t.Unix()))
return nil, nil, fmt.Errorf("incorrect timestamp %d != %d", header.Timestamp(), uint64(t.Unix()))
}

if err := validateBuilderSignature(signedBid); err != nil {
return nil, errors.Wrap(err, "could not validate builder signature")
return nil, nil, errors.Wrap(err, "could not validate builder signature")
}

var bundle *enginev1.BlindedBlobsBundle
if bid.Version() >= version.Deneb {
bundle, err = bid.BlindedBlobsBundle()
if err != nil {
return nil, nil, errors.Wrap(err, "could not get blinded blobs bundle")
}
log.WithField("blindBlobCount", len(bundle.BlobRoots))
}

log.WithFields(logrus.Fields{
Expand All @@ -223,7 +233,7 @@ func (vs *Server) getPayloadHeaderFromBuilder(ctx context.Context, slot primitiv
trace.StringAttribute("blockHash", fmt.Sprintf("%#x", header.BlockHash())),
)

return header, nil
return header, bundle, nil
}

// Validates builder signature and returns an error if the signature is invalid.
Expand Down
14 changes: 7 additions & 7 deletions beacon-chain/rpc/prysm/v1alpha1/validator/proposer_bellatrix_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ func TestServer_setExecutionData(t *testing.T) {
b := blk.Block()
localPayload, _, err := vs.getLocalPayloadAndBlobs(ctx, b, capellaTransitionState)
require.NoError(t, err)
builderPayload, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
builderPayload, _, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
require.NoError(t, err)
require.NoError(t, setExecutionData(context.Background(), blk, localPayload, builderPayload))
e, err := blk.Block().Body().Execution()
Expand Down Expand Up @@ -140,7 +140,7 @@ func TestServer_setExecutionData(t *testing.T) {

localPayload, _, err := vs.getLocalPayloadAndBlobs(ctx, b, capellaTransitionState)
require.NoError(t, err)
builderPayload, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
builderPayload, _, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
require.NoError(t, err)
require.NoError(t, setExecutionData(context.Background(), blk, localPayload, builderPayload))
e, err := blk.Block().Body().Execution()
Expand Down Expand Up @@ -202,7 +202,7 @@ func TestServer_setExecutionData(t *testing.T) {
b := blk.Block()
localPayload, _, err := vs.getLocalPayloadAndBlobs(ctx, b, capellaTransitionState)
require.NoError(t, err)
builderPayload, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
builderPayload, _, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
require.NoError(t, err)
require.NoError(t, setExecutionData(context.Background(), blk, localPayload, builderPayload))
e, err := blk.Block().Body().Execution()
Expand All @@ -216,7 +216,7 @@ func TestServer_setExecutionData(t *testing.T) {
b := blk.Block()
localPayload, _, err := vs.getLocalPayloadAndBlobs(ctx, b, capellaTransitionState)
require.NoError(t, err)
builderPayload, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
builderPayload, _, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
require.NoError(t, err)
require.NoError(t, setExecutionData(context.Background(), blk, localPayload, builderPayload))
e, err := blk.Block().Body().Execution()
Expand All @@ -236,7 +236,7 @@ func TestServer_setExecutionData(t *testing.T) {
b := blk.Block()
localPayload, _, err := vs.getLocalPayloadAndBlobs(ctx, b, capellaTransitionState)
require.NoError(t, err)
builderPayload, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
builderPayload, _, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
require.NoError(t, err)
require.NoError(t, setExecutionData(context.Background(), blk, localPayload, builderPayload))
e, err := blk.Block().Body().Execution()
Expand All @@ -257,7 +257,7 @@ func TestServer_setExecutionData(t *testing.T) {
b := blk.Block()
localPayload, _, err := vs.getLocalPayloadAndBlobs(ctx, b, capellaTransitionState)
require.NoError(t, err)
builderPayload, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
builderPayload, _, err := vs.getBuilderPayload(ctx, b.Slot(), b.ProposerIndex())
require.ErrorIs(t, consensus_types.ErrNilObjectWrapped, err) // Builder returns fault. Use local block
require.NoError(t, setExecutionData(context.Background(), blk, localPayload, builderPayload))
e, err := blk.Block().Body().Execution()
Expand Down Expand Up @@ -502,7 +502,7 @@ func TestServer_getPayloadHeader(t *testing.T) {
}}
hb, err := vs.HeadFetcher.HeadBlock(context.Background())
require.NoError(t, err)
h, err := vs.getPayloadHeaderFromBuilder(context.Background(), hb.Block().Slot(), 0)
h, _, err := vs.getPayloadHeaderFromBuilder(context.Background(), hb.Block().Slot(), 0)
if tc.err != "" {
require.ErrorContains(t, tc.err, err)
} else {
Expand Down
32 changes: 31 additions & 1 deletion beacon-chain/rpc/prysm/v1alpha1/validator/proposer_deneb.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,14 +11,19 @@ import (

// setKzgCommitments sets the KZG commitment on the block.
// Return early if the block version is older than deneb or block slot has not passed deneb epoch.
func setKzgCommitments(blk interfaces.SignedBeaconBlock, bundle *enginev1.BlobsBundle) error {
func setKzgCommitments(blk interfaces.SignedBeaconBlock, bundle *enginev1.BlobsBundle, blindBundle *enginev1.BlindedBlobsBundle) error {
if blk.Version() < version.Deneb {
return nil
}
slot := blk.Block().Slot()
if slots.ToEpoch(slot) < params.BeaconConfig().DenebForkEpoch {
return nil
}

if blk.IsBlinded() {
return blk.SetBlobKzgCommitments(blindBundle.KzgCommitments)
}

return blk.SetBlobKzgCommitments(bundle.KzgCommitments)
}

Expand Down Expand Up @@ -46,3 +51,28 @@ func blobsBundleToSidecars(bundle *enginev1.BlobsBundle, blk interfaces.ReadOnly

return sidecars, nil
}

// coverts a blinds blobs bundle to a sidecar format.
func blindBlobsBundleToSidecars(bundle *enginev1.BlindedBlobsBundle, blk interfaces.ReadOnlyBeaconBlock) ([]*ethpb.BlindedBlobSidecar, error) {
r, err := blk.HashTreeRoot()
if err != nil {
return nil, err
}
pr := blk.ParentRoot()

sidecars := make([]*ethpb.BlindedBlobSidecar, len(bundle.BlobRoots))
for i := 0; i < len(bundle.BlobRoots); i++ {
sidecars[i] = &ethpb.BlindedBlobSidecar{
BlockRoot: r[:],
Index: uint64(i),
Slot: blk.Slot(),
BlockParentRoot: pr[:],
ProposerIndex: blk.ProposerIndex(),
BlobRoot: bundle.BlobRoots[i],
KzgCommitment: bundle.KzgCommitments[i],
KzgProof: bundle.Proofs[i],
}
}

return sidecars, nil
}
6 changes: 3 additions & 3 deletions beacon-chain/rpc/prysm/v1alpha1/validator/proposer_deneb_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,18 +14,18 @@ import (
func Test_setKzgCommitments(t *testing.T) {
b, err := blocks.NewSignedBeaconBlock(util.NewBeaconBlock())
require.NoError(t, err)
require.NoError(t, setKzgCommitments(b, nil))
require.NoError(t, setKzgCommitments(b, nil, nil))
b, err = blocks.NewSignedBeaconBlock(util.NewBeaconBlockDeneb())
require.NoError(t, err)
require.NoError(t, setKzgCommitments(b, nil))
require.NoError(t, setKzgCommitments(b, nil, nil))

cfg := params.BeaconConfig().Copy()
cfg.DenebForkEpoch = 0
params.OverrideBeaconConfig(cfg)

kcs := [][]byte{[]byte("kzg"), []byte("kzg1"), []byte("kzg2")}
bundle := &enginev1.BlobsBundle{KzgCommitments: kcs}
require.NoError(t, setKzgCommitments(b, bundle))
require.NoError(t, setKzgCommitments(b, bundle, nil))
got, err := b.Block().Body().BlobKzgCommitments()
require.NoError(t, err)
require.DeepEqual(t, got, kcs)
Expand Down
8 changes: 4 additions & 4 deletions beacon-chain/rpc/prysm/v1alpha1/validator/proposer_execution_payload.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -238,20 +238,20 @@ func (vs *Server) getTerminalBlockHashIfExists(ctx context.Context, transitionTi

func (vs *Server) getBuilderPayload(ctx context.Context,
slot primitives.Slot,
vIdx primitives.ValidatorIndex) (interfaces.ExecutionData, error) {
vIdx primitives.ValidatorIndex) (interfaces.ExecutionData, *enginev1.BlindedBlobsBundle, error) {
ctx, span := trace.StartSpan(ctx, "ProposerServer.getBuilderPayload")
defer span.End()

if slots.ToEpoch(slot) < params.BeaconConfig().BellatrixForkEpoch {
return nil, nil
return nil, nil, nil
}
canUseBuilder, err := vs.canUseBuilder(ctx, slot, vIdx)
if err != nil {
return nil, errors.Wrap(err, "failed to check if we can use the builder")
return nil, nil, errors.Wrap(err, "failed to check if we can use the builder")
}
span.AddAttributes(trace.BoolAttribute("canUseBuilder", canUseBuilder))
if !canUseBuilder {
return nil, nil
return nil, nil, nil
}

return vs.getPayloadHeaderFromBuilder(ctx, slot, vIdx)
Expand Down
Loading