From e894c6bb9fd391b89674a1146bbfd77fd54038d9 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Fri, 21 Aug 2026 13:03:10 +0300 Subject: [PATCH 1/2] core: wire payload attestation into duty timing plumbing --- core/bcast/bcast.go | 2 +- core/deadline.go | 3 +++ docs/architecture.md | 1 + 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/core/bcast/bcast.go b/core/bcast/bcast.go index d8d664de5..bb353f0af 100644 --- a/core/bcast/bcast.go +++ b/core/bcast/bcast.go @@ -404,7 +404,7 @@ func newDelayFunc(ctx context.Context, eth2Cl eth2wrap.Client) (func(slot uint64 var offset time.Duration switch duty { - case core.DutyAttester, core.DutyAggregator, core.DutySyncContribution: + case core.DutyAttester, core.DutyAggregator, core.DutySyncContribution, core.DutyPayloadAttestation: offset = slotOffsetFunc(core.Duty{Slot: slot, Type: duty}) default: } diff --git a/core/deadline.go b/core/deadline.go index 61a3377f4..1cd5bc345 100644 --- a/core/deadline.go +++ b/core/deadline.go @@ -123,6 +123,9 @@ func NewDutyDeadlineFunc(ctx context.Context, eth2Cl eth2wrap.Client) (DeadlineF duration = time.Duration(slotsPerEpoch) * slotDuration case DutyPrepareAggregator, DutyPrepareSyncContribution: duration = 2 * time.Duration(slotsPerEpoch) * slotDuration + case DutyPayloadAttestation: + // Payload attestation messages are only accepted on gossip within their own slot. + duration = slotDuration default: duration = slotDuration } diff --git a/docs/architecture.md b/docs/architecture.md index ea7548154..0bbe63a89 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -136,6 +136,7 @@ type Duty struct { - `DutyPrepareSyncContribution = 11`: Prepare sync contribution duty - `DutySyncContribution = 12`: Sync contribution duty - `DutyInfoSync = 13`: Sending versions of peers in a cluster over wire +- `DutyPayloadAttestation = 14`: Payload timeliness committee (PTC) attestation duty (gloas/ePBS) > ℹ️ Duty is on a cluster level, not a DV level. A duty defines the “unit of work” for the whole cluster, > not just a single DV. This allows the workflow to aggregate and batch multiple DVs in some steps, specifically consensus. From 0a2470d675ccb9444c423fecc60a64762f3b1f95 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:53:16 +0300 Subject: [PATCH 2/2] core/validatorapi: add payload attestation endpoints --- core/interfaces.go | 8 + core/tracing.go | 9 + core/validatorapi/mocks/handler.go | 192 ++++++++++++++-------- core/validatorapi/router.go | 86 ++++++++++ core/validatorapi/router_internal_test.go | 117 +++++++++++++ core/validatorapi/validatorapi.go | 81 +++++++++ core/validatorapi/validatorapi_test.go | 61 +++++++ 7 files changed, 483 insertions(+), 71 deletions(-) diff --git a/core/interfaces.go b/core/interfaces.go index d225d8c0f..bee56e0eb 100644 --- a/core/interfaces.go +++ b/core/interfaces.go @@ -138,6 +138,9 @@ type ValidatorAPI interface { // RegisterAwaitAttestation registers a function to query attestation data. RegisterAwaitAttestation(func(ctx context.Context, slot, commIdx uint64) (*eth2p0.AttestationData, error)) + // RegisterAwaitPayloadAttestationData registers a function to query payload attestation data. + RegisterAwaitPayloadAttestationData(func(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error)) + // RegisterAwaitSyncContribution registers a function to query sync contribution data. RegisterAwaitSyncContribution(func(ctx context.Context, slot, subcommIdx uint64, beaconBlockRoot eth2p0.Root) (*altair.SyncCommitteeContribution, error)) @@ -279,6 +282,7 @@ type wireFuncs struct { DutyDBPubKeyByAttestation func(ctx context.Context, slot, commIdx, valIdx uint64) (PubKey, error) DutyDBAwaitAggAttestation func(ctx context.Context, slot uint64, attestationRoot eth2p0.Root, committeeIndex eth2p0.CommitteeIndex) (*eth2spec.VersionedAttestation, error) DutyDBAwaitSyncContribution func(ctx context.Context, slot, subcommIdx uint64, beaconBlockRoot eth2p0.Root) (*altair.SyncCommitteeContribution, error) + DutyDBAwaitPayloadAttestation func(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error) VAPIRegisterAwaitAttestation func(func(ctx context.Context, slot, commIdx uint64) (*eth2p0.AttestationData, error)) VAPIRegisterAwaitSyncContribution func(func(ctx context.Context, slot, subcommIdx uint64, beaconBlockRoot eth2p0.Root) (*altair.SyncCommitteeContribution, error)) VAPIRegisterAwaitProposal func(func(ctx context.Context, slot uint64) (*eth2api.VersionedProposal, error)) @@ -286,6 +290,7 @@ type wireFuncs struct { VAPIRegisterPubKeyByAttestation func(func(ctx context.Context, slot, commIdx, valIdx uint64) (PubKey, error)) VAPIRegisterAwaitAggAttestation func(func(ctx context.Context, slot uint64, attestationRoot eth2p0.Root, committeeIndex eth2p0.CommitteeIndex) (*eth2spec.VersionedAttestation, error)) VAPIRegisterAwaitAggSigDB func(func(context.Context, Duty, PubKey, SubcommitteeIndex) (SignedData, error)) + VAPIRegisterAwaitPayloadAttData func(func(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error)) VAPISubscribe func(func(context.Context, Duty, ParSignedDataSet) error) ParSigDBStoreInternal func(context.Context, Duty, ParSignedDataSet) error ParSigDBStoreExternal func(context.Context, Duty, ParSignedDataSet) error @@ -335,6 +340,7 @@ func Wire(sched Scheduler, DutyDBPubKeyByAttestation: dutyDB.PubKeyByAttestation, DutyDBAwaitAggAttestation: dutyDB.AwaitAggAttestation, DutyDBAwaitSyncContribution: dutyDB.AwaitSyncContribution, + DutyDBAwaitPayloadAttestation: dutyDB.AwaitPayloadAttestationData, VAPIRegisterAwaitProposal: vapi.RegisterAwaitProposal, VAPIRegisterAwaitAttestation: vapi.RegisterAwaitAttestation, VAPIRegisterAwaitSyncContribution: vapi.RegisterAwaitSyncContribution, @@ -342,6 +348,7 @@ func Wire(sched Scheduler, VAPIRegisterPubKeyByAttestation: vapi.RegisterPubKeyByAttestation, VAPIRegisterAwaitAggAttestation: vapi.RegisterAwaitAggAttestation, VAPIRegisterAwaitAggSigDB: vapi.RegisterAwaitAggSigDB, + VAPIRegisterAwaitPayloadAttData: vapi.RegisterAwaitPayloadAttestationData, VAPISubscribe: vapi.Subscribe, ParSigDBStoreInternal: parSigDB.StoreInternal, ParSigDBStoreExternal: parSigDB.StoreExternal, @@ -372,6 +379,7 @@ func Wire(sched Scheduler, w.VAPIRegisterAwaitProposal(w.DutyDBAwaitProposal) w.VAPIRegisterAwaitAttestation(w.DutyDBAwaitAttestation) w.VAPIRegisterAwaitSyncContribution(w.DutyDBAwaitSyncContribution) + w.VAPIRegisterAwaitPayloadAttData(w.DutyDBAwaitPayloadAttestation) w.VAPIRegisterGetDutyDefinition(w.SchedulerGetDutyDefinition) w.VAPIRegisterPubKeyByAttestation(w.DutyDBPubKeyByAttestation) w.VAPIRegisterAwaitAggAttestation(w.DutyDBAwaitAggAttestation) diff --git a/core/tracing.go b/core/tracing.go index 3519329a5..b801210f8 100644 --- a/core/tracing.go +++ b/core/tracing.go @@ -8,6 +8,7 @@ import ( "sync" eth2api "github.com/attestantio/go-eth2-client/api" + "github.com/attestantio/go-eth2-client/spec/gloas" "go.opentelemetry.io/otel/codes" semconv "go.opentelemetry.io/otel/semconv/v1.4.0" "go.opentelemetry.io/otel/trace" @@ -104,6 +105,14 @@ func WithTracing() WireOption { return vp, withSpanStatus(span, err) } + w.DutyDBAwaitPayloadAttestation = func(parent context.Context, slot uint64) (*gloas.PayloadAttestationData, error) { + ctx, span := tracer.Start(parent, "core/dutydb.AwaitPayloadAttestationData") + defer span.End() + + data, err := clone.DutyDBAwaitPayloadAttestation(ctx, slot) + + return data, withSpanStatus(span, err) + } w.ParSigDBStoreInternal = func(parent context.Context, duty Duty, set ParSignedDataSet) error { ctx, span := tracer.Start(parent, "core/parsigdb.StoreInternal") defer span.End() diff --git a/core/validatorapi/mocks/handler.go b/core/validatorapi/mocks/handler.go index 3e3a01525..65281d98c 100644 --- a/core/validatorapi/mocks/handler.go +++ b/core/validatorapi/mocks/handler.go @@ -1,17 +1,19 @@ // Copyright © 2022-2026 Obol Labs Inc. Licensed under the terms of a Business Source License 1.1 -// Code generated by mockery v2.53.3. DO NOT EDIT. +// Code generated by mockery v2.53.6. DO NOT EDIT. package mocks import ( - "net/http" - api "github.com/attestantio/go-eth2-client/api" altair "github.com/attestantio/go-eth2-client/spec/altair" context "context" + gloas "github.com/attestantio/go-eth2-client/spec/gloas" + + http "net/http" + mock "github.com/stretchr/testify/mock" phase0 "github.com/attestantio/go-eth2-client/spec/phase0" @@ -26,6 +28,24 @@ type Handler struct { mock.Mock } +// Address provides a mock function with no fields +func (_m *Handler) Address() string { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Address") + } + + var r0 string + if rf, ok := ret.Get(0).(func() string); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(string) + } + + return r0 +} + // AggregateAttestation provides a mock function with given fields: ctx, opts func (_m *Handler) AggregateAttestation(ctx context.Context, opts *api.AggregateAttestationOpts) (*api.Response[*spec.VersionedAttestation], error) { ret := _m.Called(ctx, opts) @@ -146,6 +166,26 @@ func (_m *Handler) BeaconCommitteeSelections(ctx context.Context, opts *api.Beac return r0, r1 } +// Headers provides a mock function with no fields +func (_m *Handler) Headers() map[string]string { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Headers") + } + + var r0 map[string]string + if rf, ok := ret.Get(0).(func() map[string]string); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(map[string]string) + } + } + + return r0 +} + // NodeVersion provides a mock function with given fields: ctx, opts func (_m *Handler) NodeVersion(ctx context.Context, opts *api.NodeVersionOpts) (*api.Response[string], error) { ret := _m.Called(ctx, opts) @@ -176,6 +216,36 @@ func (_m *Handler) NodeVersion(ctx context.Context, opts *api.NodeVersionOpts) ( return r0, r1 } +// PayloadAttestationData provides a mock function with given fields: ctx, slot +func (_m *Handler) PayloadAttestationData(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error) { + ret := _m.Called(ctx, slot) + + if len(ret) == 0 { + panic("no return value specified for PayloadAttestationData") + } + + var r0 *gloas.PayloadAttestationData + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, uint64) (*gloas.PayloadAttestationData, error)); ok { + return rf(ctx, slot) + } + if rf, ok := ret.Get(0).(func(context.Context, uint64) *gloas.PayloadAttestationData); ok { + r0 = rf(ctx, slot) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*gloas.PayloadAttestationData) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok { + r1 = rf(ctx, slot) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // Proposal provides a mock function with given fields: ctx, opts func (_m *Handler) Proposal(ctx context.Context, opts *api.ProposalOpts) (*api.Response[*api.VersionedProposal], error) { ret := _m.Called(ctx, opts) @@ -236,6 +306,36 @@ func (_m *Handler) ProposerDuties(ctx context.Context, opts *api.ProposerDutiesO return r0, r1 } +// Proxy provides a mock function with given fields: ctx, req +func (_m *Handler) Proxy(ctx context.Context, req *http.Request) (*http.Response, error) { + ret := _m.Called(ctx, req) + + if len(ret) == 0 { + panic("no return value specified for Proxy") + } + + var r0 *http.Response + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, *http.Request) (*http.Response, error)); ok { + return rf(ctx, req) + } + if rf, ok := ret.Get(0).(func(context.Context, *http.Request) *http.Response); ok { + r0 = rf(ctx, req) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*http.Response) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, *http.Request) error); ok { + r1 = rf(ctx, req) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // SubmitAggregateAttestations provides a mock function with given fields: ctx, opts func (_m *Handler) SubmitAggregateAttestations(ctx context.Context, opts *api.SubmitAggregateAttestationsOpts) error { ret := _m.Called(ctx, opts) @@ -290,6 +390,24 @@ func (_m *Handler) SubmitBlindedProposal(ctx context.Context, opts *api.SubmitBl return r0 } +// SubmitPayloadAttestationMessages provides a mock function with given fields: ctx, messages +func (_m *Handler) SubmitPayloadAttestationMessages(ctx context.Context, messages []*gloas.PayloadAttestationMessage) error { + ret := _m.Called(ctx, messages) + + if len(ret) == 0 { + panic("no return value specified for SubmitPayloadAttestationMessages") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, []*gloas.PayloadAttestationMessage) error); ok { + r0 = rf(ctx, messages) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // SubmitProposal provides a mock function with given fields: ctx, opts func (_m *Handler) SubmitProposal(ctx context.Context, opts *api.SubmitProposalOpts) error { ret := _m.Called(ctx, opts) @@ -500,74 +618,6 @@ func (_m *Handler) Validators(ctx context.Context, opts *api.ValidatorsOpts) (*a return r0, r1 } -// Proxy provides a mock function with given fields: ctx, req -func (_m *Handler) Proxy(ctx context.Context, req *http.Request) (*http.Response, error) { - ret := _m.Called(ctx, req) - - if len(ret) == 0 { - panic("no return value specified for Proxy") - } - - var r0 *http.Response - var r1 error - if rf, ok := ret.Get(0).(func(context.Context, *http.Request) (*http.Response, error)); ok { - return rf(ctx, req) - } - if rf, ok := ret.Get(0).(func(context.Context, *http.Request) *http.Response); ok { - r0 = rf(ctx, req) - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).(*http.Response) - } - } - - if rf, ok := ret.Get(1).(func(context.Context, *http.Request) error); ok { - r1 = rf(ctx, req) - } else { - r1 = ret.Error(1) - } - - return r0, r1 -} - -// Address provides a mock function with given fields: -func (_m *Handler) Address() string { - ret := _m.Called() - - if len(ret) == 0 { - panic("no return value specified for Address") - } - - var r0 string - if rf, ok := ret.Get(0).(func() string); ok { - r0 = rf() - } else { - r0 = ret.Get(0).(string) - } - - return r0 -} - -// Headers provides a mock function with given fields: -func (_m *Handler) Headers() map[string]string { - ret := _m.Called() - - if len(ret) == 0 { - panic("no return value specified for Headers") - } - - var r0 map[string]string - if rf, ok := ret.Get(0).(func() map[string]string); ok { - r0 = rf() - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).(map[string]string) - } - } - - return r0 -} - // NewHandler creates a new instance of Handler. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. // The first argument is typically a *testing.T value. func NewHandler(t interface { diff --git a/core/validatorapi/router.go b/core/validatorapi/router.go index 63a0884d5..8f4cf2f37 100644 --- a/core/validatorapi/router.go +++ b/core/validatorapi/router.go @@ -36,6 +36,7 @@ import ( "github.com/attestantio/go-eth2-client/spec/bellatrix" "github.com/attestantio/go-eth2-client/spec/capella" "github.com/attestantio/go-eth2-client/spec/electra" + "github.com/attestantio/go-eth2-client/spec/gloas" eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0" "github.com/gorilla/mux" "github.com/pk910/dynamic-ssz/sszutils" @@ -88,6 +89,15 @@ type Handler interface { eth2client.VoluntaryExitSubmitter // Above sorted alphabetically. + // TODO(gloas): replace with eth2client.PayloadAttestationDataProvider and + // eth2client.PayloadAttestationMessagesSubmitter once go-eth2-client exposes them + // (attestantio/go-eth2-client#311). + + // PayloadAttestationData returns the payload attestation data for the provided slot. + PayloadAttestationData(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error) + // SubmitPayloadAttestationMessages receives partially signed payload attestation messages. + SubmitPayloadAttestationMessages(ctx context.Context, messages []*gloas.PayloadAttestationMessage) error + // Address returns the address of the beacon node. Address() string // Headers returns custom headers to include in requests to the beacon node. @@ -288,6 +298,20 @@ func NewRouter(h Handler, builderEnabled bool) (*mux.Router, error) { Methods: []string{http.MethodPost}, Encodings: []contentType{contentTypeJSON}, }, + { + Name: "payload_attestation_data", + Path: "/eth/v1/validator/payload_attestation_data", + Handler: payloadAttestationData(h), + Methods: []string{http.MethodGet}, + Encodings: []contentType{contentTypeJSON}, + }, + { + Name: "submit_payload_attestations", + Path: "/eth/v1/beacon/pool/payload_attestations", + Handler: submitPayloadAttestationMessages(h), + Methods: []string{http.MethodPost}, + Encodings: []contentType{contentTypeJSON}, + }, { Name: "sync_committee_contribution", Path: "/eth/v1/validator/sync_committee_contribution", @@ -1581,6 +1605,68 @@ func submitAggregateAttestations(s eth2client.AggregateAttestationsSubmitter) ha } } +// payloadAttestationData returns a handler function for the payload attestation data endpoint. +func payloadAttestationData(h Handler) handlerFunc { + return func(ctx context.Context, _ map[string]string, _ http.Header, query url.Values, _ contentType, _ []byte) (any, http.Header, error) { + slot, err := uintQuery(query, "slot") + if err != nil { + return nil, nil, err + } + + data, err := h.PayloadAttestationData(ctx, slot) + if err != nil { + return nil, nil, err + } + + version := eth2spec.DataVersionGloas.String() + + return struct { + Version string `json:"version"` + Data *gloas.PayloadAttestationData `json:"data"` + }{ + Version: version, + Data: data, + }, http.Header{versionHeader: []string{version}}, nil + } +} + +// submitPayloadAttestationMessages returns a handler function for the payload attestation pool submission endpoint. +func submitPayloadAttestationMessages(h Handler) handlerFunc { + return func(ctx context.Context, _ map[string]string, header http.Header, _ url.Values, typ contentType, body []byte) (any, http.Header, error) { + var version eth2spec.DataVersion + + err := version.UnmarshalJSON([]byte("\"" + header.Get(versionHeader) + "\"")) + if err != nil { + return nil, nil, apiError{ + StatusCode: http.StatusBadRequest, + Message: "invalid or missing " + versionHeader + " header", + Err: err, + } + } + + if version != eth2spec.DataVersionGloas { + return nil, nil, apiError{ + StatusCode: http.StatusBadRequest, + Message: "unsupported " + versionHeader + " header, expected gloas", + } + } + + var msgs []*gloas.PayloadAttestationMessage + + err = unmarshal(typ, body, &msgs) + if err != nil { + return nil, nil, errors.Wrap(err, "unmarshal payload attestation messages") + } + + err = h.SubmitPayloadAttestationMessages(ctx, msgs) + if err != nil { + return nil, nil, err + } + + return nil, nil, nil + } +} + func submitSyncCommitteeMessages(s eth2client.SyncCommitteeMessagesSubmitter) handlerFunc { return func(ctx context.Context, _ map[string]string, _ http.Header, _ url.Values, typ contentType, body []byte) (any, http.Header, error) { var msgs []*altair.SyncCommitteeMessage diff --git a/core/validatorapi/router_internal_test.go b/core/validatorapi/router_internal_test.go index 29eff3d3e..629de8102 100644 --- a/core/validatorapi/router_internal_test.go +++ b/core/validatorapi/router_internal_test.go @@ -35,6 +35,7 @@ import ( "github.com/attestantio/go-eth2-client/spec/bellatrix" "github.com/attestantio/go-eth2-client/spec/capella" "github.com/attestantio/go-eth2-client/spec/electra" + "github.com/attestantio/go-eth2-client/spec/gloas" eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0" "github.com/stretchr/testify/require" @@ -2287,6 +2288,8 @@ type testHandler struct { SubmitSyncCommitteeMessagesFunc func(ctx context.Context, messages []*altair.SyncCommitteeMessage) error SyncCommitteeDutiesFunc func(ctx context.Context, opts *eth2api.SyncCommitteeDutiesOpts) (*eth2api.Response[[]*eth2v1.SyncCommitteeDuty], error) SyncCommitteeContributionFunc func(ctx context.Context, opts *eth2api.SyncCommitteeContributionOpts) (*eth2api.Response[*altair.SyncCommitteeContribution], error) + PayloadAttestationDataFunc func(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error) + SubmitPayloadAttMsgsFunc func(ctx context.Context, messages []*gloas.PayloadAttestationMessage) error ProxyFunc func(ctx context.Context, req *http.Request) (*http.Response, error) AddressFunc func() string HeadersFunc func() map[string]string @@ -2296,6 +2299,14 @@ func (h testHandler) AttestationData(ctx context.Context, opts *eth2api.Attestat return h.AttestationDataFunc(ctx, opts) } +func (h testHandler) PayloadAttestationData(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error) { + return h.PayloadAttestationDataFunc(ctx, slot) +} + +func (h testHandler) SubmitPayloadAttestationMessages(ctx context.Context, messages []*gloas.PayloadAttestationMessage) error { + return h.SubmitPayloadAttMsgsFunc(ctx, messages) +} + func (h testHandler) AttesterDuties(ctx context.Context, opts *eth2api.AttesterDutiesOpts) (*eth2api.Response[[]*eth2v1.AttesterDuty], error) { return h.AttesterDutiesFunc(ctx, opts) } @@ -2554,3 +2565,109 @@ func nest(data any, nests ...string) any { return res } + +func TestPayloadAttestationRoutes(t *testing.T) { + t.Run("payload_attestation_data", func(t *testing.T) { + expected := testutil.RandomPayloadAttestationData() + expected.Slot = 42 + + handler := testHandler{ + PayloadAttestationDataFunc: func(_ context.Context, slot uint64) (*gloas.PayloadAttestationData, error) { + require.Equal(t, uint64(42), slot) + + return expected, nil + }, + } + + callback := func(ctx context.Context, baseURL string) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL+"/eth/v1/validator/payload_attestation_data?slot=42", nil) + require.NoError(t, err) + + resp, err := new(http.Client).Do(req) + require.NoError(t, err) + + defer resp.Body.Close() + + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Equal(t, "gloas", resp.Header.Get(versionHeader)) + + var res struct { + Version string `json:"version"` + Data *gloas.PayloadAttestationData `json:"data"` + } + require.NoError(t, json.NewDecoder(resp.Body).Decode(&res)) + require.Equal(t, "gloas", res.Version) + require.Equal(t, expected, res.Data) + } + + testRawRouter(t, handler, callback) + }) + + t.Run("submit_payload_attestations", func(t *testing.T) { + msg := testutil.RandomPayloadAttestationMessage() + + var received []*gloas.PayloadAttestationMessage + + handler := testHandler{ + SubmitPayloadAttMsgsFunc: func(_ context.Context, messages []*gloas.PayloadAttestationMessage) error { + received = messages + + return nil + }, + } + + callback := func(ctx context.Context, baseURL string) { + body, err := json.Marshal([]*gloas.PayloadAttestationMessage{msg}) + require.NoError(t, err) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/eth/v1/beacon/pool/payload_attestations", bytes.NewReader(body)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + req.Header.Set(versionHeader, "gloas") + + resp, err := new(http.Client).Do(req) + require.NoError(t, err) + + defer resp.Body.Close() + + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Len(t, received, 1) + require.Equal(t, msg, received[0]) + } + + testRawRouter(t, handler, callback) + }) + + t.Run("submit_payload_attestations_bad_version_header", func(t *testing.T) { + handler := testHandler{ + SubmitPayloadAttMsgsFunc: func(_ context.Context, _ []*gloas.PayloadAttestationMessage) error { + require.Fail(t, "handler must not be called") + + return nil + }, + } + + callback := func(ctx context.Context, baseURL string) { + body, err := json.Marshal([]*gloas.PayloadAttestationMessage{testutil.RandomPayloadAttestationMessage()}) + require.NoError(t, err) + + for _, version := range []string{"", "electra"} { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/eth/v1/beacon/pool/payload_attestations", bytes.NewReader(body)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + + if version != "" { + req.Header.Set(versionHeader, version) + } + + resp, err := new(http.Client).Do(req) + require.NoError(t, err) + + require.Equal(t, http.StatusBadRequest, resp.StatusCode) + resp.Body.Close() + } + } + + testRawRouter(t, handler, callback) + }) +} diff --git a/core/validatorapi/validatorapi.go b/core/validatorapi/validatorapi.go index 31aea7573..4c54d2d7a 100644 --- a/core/validatorapi/validatorapi.go +++ b/core/validatorapi/validatorapi.go @@ -16,6 +16,7 @@ import ( eth2v1 "github.com/attestantio/go-eth2-client/api/v1" eth2spec "github.com/attestantio/go-eth2-client/spec" "github.com/attestantio/go-eth2-client/spec/altair" + "github.com/attestantio/go-eth2-client/spec/gloas" eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" @@ -189,6 +190,7 @@ type Component struct { // Registered input functions pubKeyByAttFunc func(ctx context.Context, slot, commIdx, valIdx uint64) (core.PubKey, error) awaitAttFunc func(ctx context.Context, slot, commIdx uint64) (*eth2p0.AttestationData, error) + awaitPayloadAttDataFunc func(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error) awaitProposalFunc func(ctx context.Context, slot uint64) (*eth2api.VersionedProposal, error) awaitSyncContributionFunc func(ctx context.Context, slot, subcommIdx uint64, beaconBlockRoot eth2p0.Root) (*altair.SyncCommitteeContribution, error) awaitAggAttFunc func(ctx context.Context, slot uint64, attestationRoot eth2p0.Root, committeeIndex eth2p0.CommitteeIndex) (*eth2spec.VersionedAttestation, error) @@ -209,6 +211,12 @@ func (c *Component) RegisterAwaitAttestation(fn func(ctx context.Context, slot, c.awaitAttFunc = fn } +// RegisterAwaitPayloadAttestationData registers a function to query payload attestation data. +// It only supports a single function, since it is an input of the component. +func (c *Component) RegisterAwaitPayloadAttestationData(fn func(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error)) { + c.awaitPayloadAttDataFunc = fn +} + // RegisterAwaitSyncContribution registers a function to query sync contribution data. // It only supports a single function, since it is an input of the component. func (c *Component) RegisterAwaitSyncContribution(fn func(ctx context.Context, slot, subcommIdx uint64, beaconBlockRoot eth2p0.Root) (*altair.SyncCommitteeContribution, error)) { @@ -969,6 +977,79 @@ func (c Component) SubmitSyncCommitteeMessages(ctx context.Context, messages []* return nil } +// PayloadAttestationData returns the payload attestation data for the provided slot. +func (c Component) PayloadAttestationData(ctx context.Context, slot uint64) (*gloas.PayloadAttestationData, error) { + var span trace.Span + + duty := core.NewPayloadAttestationDuty(slot) + ctx, span = core.StartDutyTrace(ctx, duty, "core/validatorapi.PayloadAttestationData") + + defer span.End() + + return c.awaitPayloadAttDataFunc(ctx, slot) +} + +// SubmitPayloadAttestationMessages receives partially signed gloas.PayloadAttestationMessage. +// - It verifies the partial signature on each message. +// - It then calls all the subscribers for further steps on the partially signed messages. +func (c Component) SubmitPayloadAttestationMessages(ctx context.Context, messages []*gloas.PayloadAttestationMessage) error { + vals, err := c.eth2Cl.ActiveValidators(ctx) + if err != nil { + return err + } + + psigsBySlot := make(map[eth2p0.Slot]core.ParSignedDataSet) + + for _, msg := range messages { + if msg.Data == nil { + return errors.New("no payload attestation data") + } + + slot := msg.Data.Slot + + eth2Pubkey, ok := vals[msg.ValidatorIndex] + if !ok { + return errors.New("validator not found") + } + + pk, err := core.PubKeyFromBytes(eth2Pubkey[:]) + if err != nil { + return err + } + + parSigData := core.NewPartialSignedPayloadAttestationMessage(msg, c.shareIdx) + + err = c.verifyPartialSig(ctx, parSigData, pk) + if err != nil { + return err + } + + log.Debug(ctx, "Payload attestation message received from validator client", + z.U64("slot", uint64(slot)), + z.Str("beacon_block_root", msg.Data.BeaconBlockRoot.String()), + z.Bool("payload_present", msg.Data.PayloadPresent)) + + _, ok = psigsBySlot[slot] + if !ok { + psigsBySlot[slot] = make(core.ParSignedDataSet) + } + + psigsBySlot[slot][pk] = parSigData + } + + for slot, data := range psigsBySlot { + duty := core.NewPayloadAttestationDuty(uint64(slot)) + for _, sub := range c.subs { + err = sub(ctx, duty, data) + if err != nil { + return err + } + } + } + + return nil +} + // SubmitSyncCommitteeContributions receives partially signed altair.SignedContributionAndProof. // - It verifies partial signature on ContributionAndProof. // - It then calls all the subscribers for further steps on partially signed contribution and proof. diff --git a/core/validatorapi/validatorapi_test.go b/core/validatorapi/validatorapi_test.go index a57018b8e..0dbcde32f 100644 --- a/core/validatorapi/validatorapi_test.go +++ b/core/validatorapi/validatorapi_test.go @@ -25,6 +25,7 @@ import ( "github.com/attestantio/go-eth2-client/spec/capella" "github.com/attestantio/go-eth2-client/spec/deneb" "github.com/attestantio/go-eth2-client/spec/electra" + "github.com/attestantio/go-eth2-client/spec/gloas" eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0" "github.com/stretchr/testify/require" @@ -2742,3 +2743,63 @@ func TestSlotFromTimestamp(t *testing.T) { }) } } + +func TestComponent_PayloadAttestationData(t *testing.T) { + ctx := context.Background() + + bmock, err := beaconmock.New(t.Context()) + require.NoError(t, err) + + vapi, err := validatorapi.NewComponentInsecure(t, bmock, 0) + require.NoError(t, err) + + expected := testutil.RandomPayloadAttestationData() + expected.Slot = 42 + + vapi.RegisterAwaitPayloadAttestationData(func(_ context.Context, slot uint64) (*gloas.PayloadAttestationData, error) { + require.Equal(t, uint64(42), slot) + + return expected, nil + }) + + data, err := vapi.PayloadAttestationData(ctx, 42) + require.NoError(t, err) + require.Equal(t, expected, data) +} + +func TestComponent_SubmitPayloadAttestationMessages(t *testing.T) { + const vIdx = 1 + + var ( + ctx = context.Background() + msg = testutil.RandomPayloadAttestationMessage() + pubkey = beaconmock.ValidatorSetA[vIdx].Validator.PublicKey + count = 0 // No of times the subscription function is called. + ) + + msg.ValidatorIndex = vIdx + + bmock, err := beaconmock.New(t.Context(), beaconmock.WithValidatorSet(beaconmock.ValidatorSetA)) + require.NoError(t, err) + + vapi, err := validatorapi.NewComponentInsecure(t, bmock, 0) + require.NoError(t, err) + + vapi.Subscribe(func(_ context.Context, duty core.Duty, set core.ParSignedDataSet) error { + require.Equal(t, core.NewPayloadAttestationDuty(uint64(msg.Data.Slot)), duty) + + pk, err := core.PubKeyFromBytes(pubkey[:]) + require.NoError(t, err) + + data, ok := set[pk] + require.True(t, ok) + require.Equal(t, core.NewPartialSignedPayloadAttestationMessage(msg, 0), data) + + count++ + + return nil + }) + + require.NoError(t, vapi.SubmitPayloadAttestationMessages(ctx, []*gloas.PayloadAttestationMessage{msg})) + require.Equal(t, 1, count) +}