diff --git a/CHANGELOG.md b/CHANGELOG.md index b8ff0077..6ff31fd7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,9 @@ gloas: - default controller.max-attestation-delay, controller.attestation-aggregation-delay, controller.max-sync-committee-message-delay and controller.sync-committee-aggregation-delay to 0; the hardcoded defaults made the spec-derived deadlines unreachable - add the payload timeliness committee duty: fetch PTC duties for the epoch, group each slot's validators into one job scheduled inside the gloas payload timing window, and vote on whether the slot's execution payload was revealed on time - submit versioned payload attestations through the immediate, multinode and null submitters, batching signatures for validators that share a slot - - obtain payload attestation data from a dedicated multiclient, configurable with strategies.payloadattestationdata.beacon-node-addresses + - obtain payload attestation data from a dedicated multiclient (the 'simple' style), configurable with strategies.payloadattestationdata.beacon-node-addresses + - add first and majority payloadattestationdata strategies, selected with strategies.payloadattestationdata.style; the majority strategy breaks a tie in favour of the payload-present vote, rejects responses that disagree on the beacon block root, and proceeds with the responses received when the timeout fires rather than discarding them + - default strategies.payloadattestationdata.timeout to 1s, tighter than the global timeout because payload attestation data is due 75% of the way through the slot; the per-style timeouts and beacon-node-addresses inherit from it - update go-eth2-client to a gloas pseudo-version - satisfy the attgo struct field order and comment capitalisation rules across services and strategies diff --git a/clients_test.go b/clients_test.go index 39118171..3a05f3e1 100644 --- a/clients_test.go +++ b/clients_test.go @@ -160,6 +160,48 @@ func TestPayloadAttesterUsesConfiguredPayloadAttestationDataProviders(t *testing require.Zero(t, globalRequests.Load()) } +func TestFirstPayloadAttestationDataStrategyRejectsInvalidResponses(t *testing.T) { + ctx := context.Background() + invalidAddress := "http://payload-data-invalid.test" + validAddress := "http://payload-data-valid.test" + invalid, err := mockconsensusclient.New(ctx) + require.NoError(t, err) + invalid.PayloadAttestationDataFunc = func(context.Context, *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + return &api.Response[*spec.VersionedPayloadAttestationData]{}, nil + } + valid, err := mockconsensusclient.New(ctx) + require.NoError(t, err) + valid.PayloadAttestationDataFunc = func(_ context.Context, opts *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + return &api.Response[*spec.VersionedPayloadAttestationData]{ + Data: &spec.VersionedPayloadAttestationData{ + Version: spec.DataVersionGloas, + Gloas: &gloas.PayloadAttestationData{Slot: opts.Slot}, + }, + }, nil + } + viper.Set("strategies.payloadattestationdata.style", "first") + viper.Set("strategies.payloadattestationdata.first.timeout", time.Second) + viper.Set("strategies.payloadattestationdata.first.beacon-node-addresses", []string{invalidAddress, validAddress}) + knownClientsMu.Lock() + knownClients[invalidAddress] = invalid + knownClients[validAddress] = valid + knownClientsMu.Unlock() + t.Cleanup(func() { + viper.Reset() + knownClientsMu.Lock() + delete(knownClients, invalidAddress) + delete(knownClients, validAddress) + knownClientsMu.Unlock() + }) + + provider, err := selectPayloadAttestationDataProvider(ctx, null.New()) + require.NoError(t, err) + response, err := provider.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + require.NoError(t, err) + require.NotNil(t, response.Data) + require.Equal(t, phase0.Slot(12), response.Data.Gloas.Slot) +} + type payloadAttestationDataSigner struct{} func (*payloadAttestationDataSigner) SignPayloadAttestationData(_ context.Context, accounts []e2wtypes.Account, _ *gloas.PayloadAttestationData) ([]phase0.BLSSignature, error) { diff --git a/docs/configuration.md b/docs/configuration.md index 80db5b26..c7db0f1a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -219,11 +219,23 @@ strategies: # bid-gap is the gap between receiving a response from a relay and querying it again. bid-gap: '100ms' # The payloadattestationdata strategy obtains payload attestation data for payload timeliness committee duties. - # It uses a single multiclient; there is no style to select. payloadattestationdata: + # style can be 'simple', 'first', or 'majority'. Unknown or absent values use 'simple'. + style: 'simple' # beacon-node-addresses are the addresses from which to receive payload attestation data, falling back to the # top-level beacon-node-addresses if not set. beacon-node-addresses: ['localhost:4000', 'localhost:5051', 'localhost:5052'] + first: + # beacon-node-addresses and timeout fall back to payloadattestationdata and then the top-level setting. + beacon-node-addresses: ['localhost:4000', 'localhost:5051', 'localhost:5052'] + timeout: '1s' + majority: + # beacon-node-addresses and timeout fall back to payloadattestationdata and then the top-level setting. + beacon-node-addresses: ['localhost:4000', 'localhost:5051', 'localhost:5052'] + timeout: '1s' + # threshold is the minimum number of matching valid responses. 0 is valid. + # At the timeout the strategy proceeds with the responses it has received. + threshold: 0 # The signedbeaconblock strategy obtains the signed beacon blocks from multiple beacon nodes. signedbeaconblock: # style can be 'first'. If not defined, the 'first' style will be used. diff --git a/main.go b/main.go index 00b25d87..20101093 100644 --- a/main.go +++ b/main.go @@ -98,6 +98,8 @@ import ( "github.com/attestantio/vouch/strategies/builderbid" bestbuilderbidstrategy "github.com/attestantio/vouch/strategies/builderbid/best" deadlinebuilderbidstrategy "github.com/attestantio/vouch/strategies/builderbid/deadline" + firstpayloadattestationdatastrategy "github.com/attestantio/vouch/strategies/payloadattestationdata/first" + majoritypayloadattestationdatastrategy "github.com/attestantio/vouch/strategies/payloadattestationdata/majority" firstsignedbeaconblockstrategy "github.com/attestantio/vouch/strategies/signedbeaconblock/first" bestsynccommitteecontributionstrategy "github.com/attestantio/vouch/strategies/synccommitteecontribution/best" firstsynccommitteecontributionstrategy "github.com/attestantio/vouch/strategies/synccommitteecontribution/first" @@ -274,6 +276,9 @@ func fetchConfig() error { viper.SetDefault("beaconblockproposer.builder-boost-factor", 91) viper.SetDefault("strategies.builderbid.deadline.deadline", time.Second) viper.SetDefault("strategies.builderbid.deadline.bid-gap", 100*time.Millisecond) + // Payload attestation data is due 75% of the way through the slot, so default tighter than the + // global timeout. Set at the strategy level so that the per-style timeouts still inherit from it. + viper.SetDefault("strategies.payloadattestationdata.timeout", time.Second) viper.SetDefault("submitter.style", "multinode") viper.SetDefault("multiinstance.static-delay.attester-delay", time.Second) viper.SetDefault("multiinstance.static-delay.proposer-delay", 2*time.Second) @@ -1698,15 +1703,11 @@ func startPayloadAttester(ctx context.Context, if !ok { return nil, nil } - payloadAttestationDataClient, err := fetchMultiClient(ctx, monitor, - "payloadattestationdata", - util.BeaconNodeAddressesForPayloadAttestationData(), - ) + payloadAttestationDataProvider, err := selectPayloadAttestationDataProvider(ctx, monitor) if err != nil { - return nil, errors.Wrap(err, "failed to fetch clients for payload attestation data") + return nil, errors.Wrap(err, "failed to obtain payload attestation data provider") } - payloadAttestationDataProvider, ok := payloadAttestationDataClient.(eth2client.PayloadAttestationDataProvider) - if !ok { + if payloadAttestationDataProvider == nil { return nil, nil } @@ -1725,6 +1726,59 @@ func startPayloadAttester(ctx context.Context, return service, nil } +func selectPayloadAttestationDataProvider(ctx context.Context, monitor metrics.Service) (eth2client.PayloadAttestationDataProvider, error) { + switch viper.GetString("strategies.payloadattestationdata.style") { + case "first": + providers, err := genericAddressToClientMapper[eth2client.PayloadAttestationDataProvider](ctx, monitor, + "strategies.payloadattestationdata.first", + "first payload attestation data strategy") + if err != nil { + return nil, err + } + provider, err := firstpayloadattestationdatastrategy.New(ctx, + firstpayloadattestationdatastrategy.WithClientMonitor(monitor.(metrics.ClientMonitor)), + firstpayloadattestationdatastrategy.WithLogLevel(util.LogLevel("strategies.payloadattestationdata.first")), + firstpayloadattestationdatastrategy.WithPayloadAttestationDataProviders(providers), + firstpayloadattestationdatastrategy.WithTimeout(util.Timeout("strategies.payloadattestationdata.first")), + ) + if err != nil { + return nil, errors.Wrap(err, "failed to start first payload attestation data strategy") + } + return provider, nil + case "majority": + providers, err := genericAddressToClientMapper[eth2client.PayloadAttestationDataProvider](ctx, monitor, + "strategies.payloadattestationdata.majority", + "majority payload attestation data strategy") + if err != nil { + return nil, err + } + provider, err := majoritypayloadattestationdatastrategy.New(ctx, + majoritypayloadattestationdatastrategy.WithClientMonitor(monitor.(metrics.ClientMonitor)), + majoritypayloadattestationdatastrategy.WithLogLevel(util.LogLevel("strategies.payloadattestationdata.majority")), + majoritypayloadattestationdatastrategy.WithPayloadAttestationDataProviders(providers), + majoritypayloadattestationdatastrategy.WithTimeout(util.Timeout("strategies.payloadattestationdata.majority")), + majoritypayloadattestationdatastrategy.WithThreshold(viper.GetInt("strategies.payloadattestationdata.majority.threshold")), + ) + if err != nil { + return nil, errors.Wrap(err, "failed to start majority payload attestation data strategy") + } + return provider, nil + default: + payloadAttestationDataClient, err := fetchMultiClient(ctx, monitor, + "payloadattestationdata", + util.BeaconNodeAddressesForPayloadAttestationData(), + ) + if err != nil { + return nil, err + } + provider, ok := payloadAttestationDataClient.(eth2client.PayloadAttestationDataProvider) + if !ok { + return nil, nil + } + return provider, nil + } +} + func genericAddressToClientMapper[T any](ctx context.Context, monitor metrics.Service, path, description string) (map[string]T, error) { addressToClientMap := make(map[string]T) for _, address := range util.BeaconNodeAddresses(path) { diff --git a/strategies/payloadattestationdata/first/parameters.go b/strategies/payloadattestationdata/first/parameters.go new file mode 100644 index 00000000..3f002e5f --- /dev/null +++ b/strategies/payloadattestationdata/first/parameters.go @@ -0,0 +1,86 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package first obtains payload attestation data from multiple nodes and selects the first valid response. +package first + +import ( + "time" + + eth2client "github.com/attestantio/go-eth2-client" + "github.com/attestantio/vouch/services/metrics" + nullmetrics "github.com/attestantio/vouch/services/metrics/null" + "github.com/pkg/errors" + "github.com/rs/zerolog" +) + +type parameters struct { + logLevel zerolog.Level + clientMonitor metrics.ClientMonitor + payloadAttestationDataProviders map[string]eth2client.PayloadAttestationDataProvider + timeout time.Duration +} + +// Parameter is the interface for service parameters. +type Parameter interface { + apply(parameters *parameters) +} + +type parameterFunc func(*parameters) + +func (f parameterFunc) apply(parameters *parameters) { + f(parameters) +} + +// WithLogLevel sets the log level for the module. +func WithLogLevel(logLevel zerolog.Level) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.logLevel = logLevel }) +} + +// WithClientMonitor sets the client monitor for the service. +func WithClientMonitor(monitor metrics.ClientMonitor) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.clientMonitor = monitor }) +} + +// WithPayloadAttestationDataProviders sets the payload attestation data providers. +func WithPayloadAttestationDataProviders(providers map[string]eth2client.PayloadAttestationDataProvider) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.payloadAttestationDataProviders = providers }) +} + +// WithTimeout sets the timeout for requests. +func WithTimeout(timeout time.Duration) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.timeout = timeout }) +} + +func parseAndCheckParameters(params ...Parameter) (*parameters, error) { + parameters := ¶meters{ + logLevel: zerolog.GlobalLevel(), + clientMonitor: nullmetrics.New(), + timeout: time.Second, + } + for _, param := range params { + if param != nil { + param.apply(parameters) + } + } + if parameters.clientMonitor == nil { + return nil, errors.New("no client monitor specified") + } + if len(parameters.payloadAttestationDataProviders) == 0 { + return nil, errors.New("no payload attestation data providers specified") + } + if parameters.timeout <= 0 { + return nil, errors.New("timeout must be positive") + } + return parameters, nil +} diff --git a/strategies/payloadattestationdata/first/service.go b/strategies/payloadattestationdata/first/service.go new file mode 100644 index 00000000..ba9ecc6e --- /dev/null +++ b/strategies/payloadattestationdata/first/service.go @@ -0,0 +1,125 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package first + +import ( + "context" + "time" + + eth2client "github.com/attestantio/go-eth2-client" + "github.com/attestantio/go-eth2-client/api" + "github.com/attestantio/go-eth2-client/spec" + "github.com/attestantio/vouch/services/metrics" + "github.com/pkg/errors" + "github.com/rs/zerolog" + zerologger "github.com/rs/zerolog/log" +) + +// Service provides payload attestation data. +type Service struct { + log zerolog.Logger + clientMonitor metrics.ClientMonitor + payloadAttestationDataProviders map[string]eth2client.PayloadAttestationDataProvider + timeout time.Duration +} + +type payloadAttestationDataResult struct { + provider string + response *api.Response[*spec.VersionedPayloadAttestationData] + err error +} + +// New creates a payload attestation data strategy. +func New(_ context.Context, params ...Parameter) (*Service, error) { + parameters, err := parseAndCheckParameters(params...) + if err != nil { + return nil, errors.Wrap(err, "problem with parameters") + } + log := zerologger.With().Str("strategy", "payloadattestationdata").Str("impl", "first").Logger() + if parameters.logLevel != log.GetLevel() { + log = log.Level(parameters.logLevel) + } + return &Service{ + log: log, + clientMonitor: parameters.clientMonitor, + payloadAttestationDataProviders: parameters.payloadAttestationDataProviders, + timeout: parameters.timeout, + }, nil +} + +// PayloadAttestationData obtains the first valid payload attestation data response. +func (s *Service) PayloadAttestationData(ctx context.Context, opts *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + ctx, cancel := context.WithTimeout(ctx, s.timeout) + defer cancel() + + started := time.Now() + results := s.issuePayloadAttestationDataRequests(ctx, opts, started) + + for range s.payloadAttestationDataProviders { + select { + case <-ctx.Done(): + // Deadline reached; prefer a response that already arrived over the expired context. + for len(results) > 0 { + if response := s.selectedPayloadAttestationData(opts, <-results, started); response != nil { + return response, nil + } + } + + return nil, errors.Wrap(ctx.Err(), "failed to obtain payload attestation data") + case result := <-results: + if response := s.selectedPayloadAttestationData(opts, result, started); response != nil { + return response, nil + } + } + } + + if ctx.Err() != nil { + return nil, errors.Wrap(ctx.Err(), "failed to obtain payload attestation data") + } + + return nil, errors.New("no valid payload attestation data received") +} + +func (s *Service) issuePayloadAttestationDataRequests(ctx context.Context, opts *api.PayloadAttestationDataOpts, started time.Time) <-chan payloadAttestationDataResult { + results := make(chan payloadAttestationDataResult, len(s.payloadAttestationDataProviders)) + for name, provider := range s.payloadAttestationDataProviders { + go func(providerName string, provider eth2client.PayloadAttestationDataProvider) { + response, err := provider.PayloadAttestationData(ctx, opts) + s.clientMonitor.ClientOperation(providerName, "payload attestation data", err == nil, time.Since(started)) + results <- payloadAttestationDataResult{provider: providerName, response: response, err: err} + }(name, provider) + } + + return results +} + +// selectedPayloadAttestationData returns the response if the result is valid for the request, otherwise nil. +func (s *Service) selectedPayloadAttestationData(opts *api.PayloadAttestationDataOpts, result payloadAttestationDataResult, started time.Time) *api.Response[*spec.VersionedPayloadAttestationData] { + if result.err != nil { + s.log.Debug().Err(result.err).Str("provider", result.provider).Msg("Failed to obtain payload attestation data") + + return nil + } + if result.response == nil || result.response.Data == nil || result.response.Data.Version != spec.DataVersionGloas || result.response.Data.Gloas == nil || result.response.Data.Gloas.Slot != opts.Slot { + s.log.Debug().Str("provider", result.provider).Msg("Received invalid payload attestation data") + + return nil + } + s.clientMonitor.StrategyOperation("first", result.provider, "payload attestation data", time.Since(started)) + + return &api.Response[*spec.VersionedPayloadAttestationData]{ + Data: result.response.Data, + Metadata: make(map[string]any), + } +} diff --git a/strategies/payloadattestationdata/first/service_test.go b/strategies/payloadattestationdata/first/service_test.go new file mode 100644 index 00000000..09b94591 --- /dev/null +++ b/strategies/payloadattestationdata/first/service_test.go @@ -0,0 +1,110 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package first_test + +import ( + "context" + "sync" + "testing" + "time" + + eth2client "github.com/attestantio/go-eth2-client" + "github.com/attestantio/go-eth2-client/api" + mockclient "github.com/attestantio/go-eth2-client/mock" + "github.com/attestantio/go-eth2-client/spec" + "github.com/attestantio/go-eth2-client/spec/gloas" + "github.com/attestantio/go-eth2-client/spec/phase0" + "github.com/attestantio/vouch/strategies/payloadattestationdata/first" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +type monitor struct { + mu sync.Mutex + clients []string + strategy []string +} + +func (m *monitor) ClientOperation(provider string, _ string, _ bool, _ time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() + m.clients = append(m.clients, provider) +} + +func (m *monitor) StrategyOperation(_ string, provider string, _ string, _ time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() + m.strategy = append(m.strategy, provider) +} + +func payloadAttestationDataProvider(t *testing.T, fn func(context.Context, *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error)) eth2client.PayloadAttestationDataProvider { + t.Helper() + provider, err := mockclient.New(context.Background()) + require.NoError(t, err) + provider.PayloadAttestationDataFunc = fn + return provider +} + +func validData(slot phase0.Slot) *api.Response[*spec.VersionedPayloadAttestationData] { + return &api.Response[*spec.VersionedPayloadAttestationData]{ + Data: &spec.VersionedPayloadAttestationData{ + Version: spec.DataVersionGloas, + Gloas: &gloas.PayloadAttestationData{ + Slot: slot, + PayloadPresent: true, + }, + }, + } +} + +func TestPayloadAttestationData(t *testing.T) { + ctx := context.Background() + monitor := &monitor{} + service, err := first.New(ctx, + first.WithLogLevel(zerolog.Disabled), + first.WithClientMonitor(monitor), + first.WithPayloadAttestationDataProviders(map[string]eth2client.PayloadAttestationDataProvider{ + "invalid": payloadAttestationDataProvider(t, func(context.Context, *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + return &api.Response[*spec.VersionedPayloadAttestationData]{}, nil + }), + "valid": payloadAttestationDataProvider(t, func(_ context.Context, opts *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + time.Sleep(10 * time.Millisecond) + return validData(opts.Slot), nil + }), + }), + ) + require.NoError(t, err) + + response, err := service.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + require.NoError(t, err) + require.Equal(t, validData(12).Data, response.Data) + require.ElementsMatch(t, []string{"invalid", "valid"}, monitor.clients) + require.Equal(t, []string{"valid"}, monitor.strategy) +} + +func TestPayloadAttestationDataHonoursCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + service, err := first.New(ctx, + first.WithPayloadAttestationDataProviders(map[string]eth2client.PayloadAttestationDataProvider{ + "slow": payloadAttestationDataProvider(t, func(ctx context.Context, _ *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + <-ctx.Done() + return nil, ctx.Err() + }), + }), + ) + require.NoError(t, err) + _, err = service.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + require.ErrorIs(t, err, context.Canceled) +} diff --git a/strategies/payloadattestationdata/majority/parameters.go b/strategies/payloadattestationdata/majority/parameters.go new file mode 100644 index 00000000..b3c3fd18 --- /dev/null +++ b/strategies/payloadattestationdata/majority/parameters.go @@ -0,0 +1,95 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package majority obtains payload attestation data from multiple nodes and selects the strongest agreement. +package majority + +import ( + "time" + + eth2client "github.com/attestantio/go-eth2-client" + "github.com/attestantio/vouch/services/metrics" + nullmetrics "github.com/attestantio/vouch/services/metrics/null" + "github.com/pkg/errors" + "github.com/rs/zerolog" +) + +type parameters struct { + logLevel zerolog.Level + clientMonitor metrics.ClientMonitor + payloadAttestationDataProviders map[string]eth2client.PayloadAttestationDataProvider + timeout time.Duration + threshold int +} + +// Parameter is the interface for service parameters. +type Parameter interface { + apply(parameters *parameters) +} + +type parameterFunc func(*parameters) + +func (f parameterFunc) apply(parameters *parameters) { + f(parameters) +} + +// WithLogLevel sets the log level for the module. +func WithLogLevel(logLevel zerolog.Level) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.logLevel = logLevel }) +} + +// WithClientMonitor sets the client monitor for the service. +func WithClientMonitor(monitor metrics.ClientMonitor) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.clientMonitor = monitor }) +} + +// WithPayloadAttestationDataProviders sets the payload attestation data providers. +func WithPayloadAttestationDataProviders(providers map[string]eth2client.PayloadAttestationDataProvider) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.payloadAttestationDataProviders = providers }) +} + +// WithTimeout sets the timeout for requests. +func WithTimeout(timeout time.Duration) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.timeout = timeout }) +} + +// WithThreshold sets the minimum number of matching valid responses. +func WithThreshold(threshold int) Parameter { + return parameterFunc(func(parameters *parameters) { parameters.threshold = threshold }) +} + +func parseAndCheckParameters(params ...Parameter) (*parameters, error) { + parameters := ¶meters{ + logLevel: zerolog.GlobalLevel(), + clientMonitor: nullmetrics.New(), + timeout: time.Second, + } + for _, param := range params { + if param != nil { + param.apply(parameters) + } + } + if parameters.clientMonitor == nil { + return nil, errors.New("no client monitor specified") + } + if len(parameters.payloadAttestationDataProviders) == 0 { + return nil, errors.New("no payload attestation data providers specified") + } + if parameters.timeout <= 0 { + return nil, errors.New("timeout must be positive") + } + if parameters.threshold < 0 { + return nil, errors.New("threshold cannot be negative") + } + return parameters, nil +} diff --git a/strategies/payloadattestationdata/majority/service.go b/strategies/payloadattestationdata/majority/service.go new file mode 100644 index 00000000..1b006175 --- /dev/null +++ b/strategies/payloadattestationdata/majority/service.go @@ -0,0 +1,215 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package majority + +import ( + "context" + "time" + + eth2client "github.com/attestantio/go-eth2-client" + "github.com/attestantio/go-eth2-client/api" + "github.com/attestantio/go-eth2-client/spec" + "github.com/attestantio/go-eth2-client/spec/phase0" + "github.com/attestantio/vouch/services/metrics" + "github.com/pkg/errors" + "github.com/rs/zerolog" + zerologger "github.com/rs/zerolog/log" +) + +// Service provides payload attestation data. +type Service struct { + log zerolog.Logger + clientMonitor metrics.ClientMonitor + payloadAttestationDataProviders map[string]eth2client.PayloadAttestationDataProvider + timeout time.Duration + threshold int +} + +// New creates a payload attestation data strategy. +func New(_ context.Context, params ...Parameter) (*Service, error) { + parameters, err := parseAndCheckParameters(params...) + if err != nil { + return nil, errors.Wrap(err, "problem with parameters") + } + log := zerologger.With().Str("strategy", "payloadattestationdata").Str("impl", "majority").Logger() + if parameters.logLevel != log.GetLevel() { + log = log.Level(parameters.logLevel) + } + return &Service{ + log: log, + clientMonitor: parameters.clientMonitor, + payloadAttestationDataProviders: parameters.payloadAttestationDataProviders, + timeout: parameters.timeout, + threshold: parameters.threshold, + }, nil +} + +type payloadAttestationDataKey struct { + version spec.DataVersion + beaconBlockRoot phase0.Root + slot phase0.Slot + payloadPresent bool + blobDataAvailable bool +} + +type payloadAttestationDataResult struct { + provider string + response *api.Response[*spec.VersionedPayloadAttestationData] + err error +} + +// PayloadAttestationData obtains the strongest valid agreement for payload attestation data. +func (s *Service) PayloadAttestationData(ctx context.Context, opts *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + ctx, cancel := context.WithTimeout(ctx, s.timeout) + defer cancel() + + started := time.Now() + buckets := s.payloadAttestationDataBuckets(ctx, opts, started) + + return s.selectPayloadAttestationData(ctx, started, buckets) +} + +func (s *Service) payloadAttestationDataBuckets(ctx context.Context, opts *api.PayloadAttestationDataOpts, started time.Time) map[payloadAttestationDataKey][]payloadAttestationDataResult { + results := s.issuePayloadAttestationDataRequests(ctx, opts, started) + buckets := make(map[payloadAttestationDataKey][]payloadAttestationDataResult) + requiredCount := max(len(s.payloadAttestationDataProviders)/2+1, s.threshold) + largestCount := 0 + bucket := func(result payloadAttestationDataResult) { + if key, ok := s.payloadAttestationDataKey(opts, result); ok { + buckets[key] = append(buckets[key], result) + largestCount = max(largestCount, len(buckets[key])) + } + } + for range s.payloadAttestationDataProviders { + if largestCount >= requiredCount { + break + } + select { + case <-ctx.Done(): + // Deadline reached; consider the outstanding providers timed out and + // proceed with the responses that did arrive. + for len(results) > 0 { + bucket(<-results) + } + s.log.Debug().Int("buckets", len(buckets)).Msg("Timed out awaiting payload attestation data") + + return buckets + case result := <-results: + bucket(result) + } + } + + return buckets +} + +func (s *Service) issuePayloadAttestationDataRequests(ctx context.Context, opts *api.PayloadAttestationDataOpts, started time.Time) <-chan payloadAttestationDataResult { + results := make(chan payloadAttestationDataResult, len(s.payloadAttestationDataProviders)) + for name, provider := range s.payloadAttestationDataProviders { + go func(providerName string, provider eth2client.PayloadAttestationDataProvider) { + response, err := provider.PayloadAttestationData(ctx, opts) + s.clientMonitor.ClientOperation(providerName, "payload attestation data", err == nil, time.Since(started)) + results <- payloadAttestationDataResult{provider: providerName, response: response, err: err} + }(name, provider) + } + return results +} + +func (s *Service) payloadAttestationDataKey(opts *api.PayloadAttestationDataOpts, result payloadAttestationDataResult) (payloadAttestationDataKey, bool) { + if result.err != nil { + s.log.Debug().Err(result.err).Str("provider", result.provider).Msg("Failed to obtain payload attestation data") + return payloadAttestationDataKey{}, false + } + if result.response == nil || result.response.Data == nil || result.response.Data.Version != spec.DataVersionGloas || result.response.Data.Gloas == nil || result.response.Data.Gloas.Slot != opts.Slot { + s.log.Debug().Str("provider", result.provider).Msg("Received invalid payload attestation data") + return payloadAttestationDataKey{}, false + } + return payloadAttestationDataKey{ + version: result.response.Data.Version, + beaconBlockRoot: result.response.Data.Gloas.BeaconBlockRoot, + slot: result.response.Data.Gloas.Slot, + payloadPresent: result.response.Data.Gloas.PayloadPresent, + blobDataAvailable: result.response.Data.Gloas.BlobDataAvailable, + }, true +} + +func (s *Service) selectPayloadAttestationData(ctx context.Context, started time.Time, buckets map[payloadAttestationDataKey][]payloadAttestationDataResult) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + leading, count := leadingPayloadAttestationDataBuckets(buckets) + if count == 0 { + if ctx.Err() != nil { + return nil, errors.Wrap(ctx.Err(), "failed to obtain payload attestation data") + } + + return nil, errors.New("no valid payload attestation data received") + } + if count < s.threshold { + s.log.Debug().Int("count", count).Int("threshold", s.threshold).Msg("Insufficient payload attestation data agreement") + return nil, errors.Errorf("payload attestation data count of %d lower than threshold %d", count, s.threshold) + } + if len(leading) == 1 { + return s.selectedPayloadAttestationData(started, buckets[leading[0]]), nil + } + return s.resolvePayloadAttestationDataTie(started, leading, buckets) +} + +func leadingPayloadAttestationDataBuckets(buckets map[payloadAttestationDataKey][]payloadAttestationDataResult) ([]payloadAttestationDataKey, int) { + leading := make([]payloadAttestationDataKey, 0, 2) + count := 0 + for key, responses := range buckets { + switch { + case len(responses) > count: + count = len(responses) + leading = append(leading[:0], key) + case len(responses) == count: + leading = append(leading, key) + } + } + return leading, count +} + +func (s *Service) resolvePayloadAttestationDataTie(started time.Time, leading []payloadAttestationDataKey, buckets map[payloadAttestationDataKey][]payloadAttestationDataResult) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + first := leading[0] + for _, key := range leading[1:] { + if key.version != first.version || key.slot != first.slot || key.beaconBlockRoot != first.beaconBlockRoot { + s.log.Debug().Msg("Split payload attestation data roots") + return nil, errors.New("split-root payload attestation data responses") + } + } + + var selected *payloadAttestationDataKey + for i := range leading { + if leading[i].payloadPresent { + if selected != nil { + s.log.Debug().Msg("Split payload attestation data responses") + return nil, errors.New("split-response payload attestation data responses") + } + selected = &leading[i] + } + } + if selected == nil { + s.log.Debug().Msg("Split payload attestation data responses") + return nil, errors.New("split-response payload attestation data responses") + } + s.log.Debug().Msg("Resolved payload attestation data tie with payload present") + return s.selectedPayloadAttestationData(started, buckets[*selected]), nil +} + +func (s *Service) selectedPayloadAttestationData(started time.Time, responses []payloadAttestationDataResult) *api.Response[*spec.VersionedPayloadAttestationData] { + for _, response := range responses { + s.clientMonitor.StrategyOperation("majority", response.provider, "payload attestation data", time.Since(started)) + } + return &api.Response[*spec.VersionedPayloadAttestationData]{ + Data: responses[0].response.Data, + Metadata: make(map[string]any), + } +} diff --git a/strategies/payloadattestationdata/majority/service_test.go b/strategies/payloadattestationdata/majority/service_test.go new file mode 100644 index 00000000..70f77608 --- /dev/null +++ b/strategies/payloadattestationdata/majority/service_test.go @@ -0,0 +1,293 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package majority_test + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + eth2client "github.com/attestantio/go-eth2-client" + "github.com/attestantio/go-eth2-client/api" + mockclient "github.com/attestantio/go-eth2-client/mock" + "github.com/attestantio/go-eth2-client/spec" + "github.com/attestantio/go-eth2-client/spec/gloas" + "github.com/attestantio/go-eth2-client/spec/phase0" + "github.com/attestantio/vouch/strategies/payloadattestationdata/majority" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +type monitor struct { + mu sync.Mutex + clients []string + strategy []string +} + +func (m *monitor) ClientOperation(provider string, _ string, _ bool, _ time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() + m.clients = append(m.clients, provider) +} + +func (m *monitor) StrategyOperation(_ string, provider string, _ string, _ time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() + m.strategy = append(m.strategy, provider) +} + +func provider(t *testing.T, response *api.Response[*spec.VersionedPayloadAttestationData]) eth2client.PayloadAttestationDataProvider { + t.Helper() + provider, err := mockclient.New(context.Background()) + require.NoError(t, err) + provider.PayloadAttestationDataFunc = func(context.Context, *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + return response, nil + } + return provider +} + +func data(slot phase0.Slot, root phase0.Root, payloadPresent, blobDataAvailable bool) *api.Response[*spec.VersionedPayloadAttestationData] { + return &api.Response[*spec.VersionedPayloadAttestationData]{ + Data: &spec.VersionedPayloadAttestationData{ + Version: spec.DataVersionGloas, + Gloas: &gloas.PayloadAttestationData{ + Slot: slot, + BeaconBlockRoot: root, + PayloadPresent: payloadPresent, + BlobDataAvailable: blobDataAvailable, + }, + }, + } +} + +func erroringProvider(t *testing.T, err error) eth2client.PayloadAttestationDataProvider { + t.Helper() + provider, newErr := mockclient.New(context.Background()) + require.NoError(t, newErr) + provider.PayloadAttestationDataFunc = func(context.Context, *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + return nil, err + } + return provider +} + +func TestPayloadAttestationDataAgreement(t *testing.T) { + ctx := context.Background() + response := data(12, phase0.Root{1}, true, true) + monitor := &monitor{} + service, err := majority.New(ctx, + majority.WithLogLevel(zerolog.Disabled), + majority.WithClientMonitor(monitor), + majority.WithTimeout(time.Second), + majority.WithThreshold(2), + majority.WithPayloadAttestationDataProviders(map[string]eth2client.PayloadAttestationDataProvider{ + "one": provider(t, response), + "two": provider(t, response), + }), + ) + require.NoError(t, err) + + actual, err := service.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + require.NoError(t, err) + require.Equal(t, response.Data, actual.Data) + require.Empty(t, actual.Metadata) + require.ElementsMatch(t, []string{"one", "two"}, monitor.clients) + require.ElementsMatch(t, []string{"one", "two"}, monitor.strategy) +} + +func delayedProvider(t *testing.T, response *api.Response[*spec.VersionedPayloadAttestationData], delay time.Duration) eth2client.PayloadAttestationDataProvider { + t.Helper() + provider, err := mockclient.New(context.Background()) + require.NoError(t, err) + provider.PayloadAttestationDataFunc = func(ctx context.Context, _ *api.PayloadAttestationDataOpts) (*api.Response[*spec.VersionedPayloadAttestationData], error) { + select { + case <-time.After(delay): + return response, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return provider +} + +func TestPayloadAttestationDataThresholdExceedingStrictMajority(t *testing.T) { + ctx := context.Background() + root := phase0.Root{1} + response := data(12, root, true, true) + service, err := majority.New(ctx, + majority.WithTimeout(time.Second), + majority.WithThreshold(4), + majority.WithPayloadAttestationDataProviders(map[string]eth2client.PayloadAttestationDataProvider{ + "one": provider(t, response), + "two": provider(t, response), + "three": provider(t, response), + "four": delayedProvider(t, response, 10*time.Millisecond), + "other": provider(t, data(12, phase0.Root{2}, true, true)), + }), + ) + require.NoError(t, err) + actual, err := service.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + require.NoError(t, err) + require.Equal(t, response.Data, actual.Data) +} + +func TestPayloadAttestationDataUsesResponsesReceivedAtTimeout(t *testing.T) { + ctx := context.Background() + root := phase0.Root{1} + response := data(12, root, true, true) + service, err := majority.New(ctx, + majority.WithLogLevel(zerolog.Disabled), + majority.WithTimeout(100*time.Millisecond), + majority.WithPayloadAttestationDataProviders(map[string]eth2client.PayloadAttestationDataProvider{ + "one": provider(t, response), + "two": provider(t, response), + "three": provider(t, data(12, phase0.Root{2}, true, true)), + // A strict majority of 3 is never reached, so the strategy waits for this + // provider until the timeout and must then use the responses it has. + "hung": delayedProvider(t, response, time.Minute), + }), + ) + require.NoError(t, err) + + actual, err := service.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + require.NoError(t, err) + require.Equal(t, response.Data, actual.Data) +} + +func TestPayloadAttestationDataSelection(t *testing.T) { + ctx := context.Background() + root1 := phase0.Root{1} + root2 := phase0.Root{2} + tests := []struct { + name string + providers map[string]eth2client.PayloadAttestationDataProvider + threshold int + expected *spec.VersionedPayloadAttestationData + err string + }{ + { + name: "UniquePlurality", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "one": provider(t, data(12, root1, true, true)), + "two": provider(t, data(12, root1, true, true)), + "three": provider(t, data(12, root2, true, true)), + "four": provider(t, data(12, phase0.Root{3}, true, true)), + }, + threshold: 2, + expected: data(12, root1, true, true).Data, + }, + { + name: "ThresholdFailure", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "one": provider(t, data(12, root1, true, true)), + "two": provider(t, data(12, root1, true, true)), + }, + threshold: 3, + err: "payload attestation data count of 2 lower than threshold 3", + }, + { + name: "ThresholdZero", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "one": provider(t, data(12, root1, true, true)), + }, + expected: data(12, root1, true, true).Data, + }, + { + name: "NoValidResponse", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "invalid": provider(t, &api.Response[*spec.VersionedPayloadAttestationData]{}), + }, + err: "no valid payload attestation data received", + }, + { + name: "InvalidResponseDoesNotVote", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "invalid": provider(t, &api.Response[*spec.VersionedPayloadAttestationData]{}), + "valid": provider(t, data(12, root1, true, true)), + }, + threshold: 1, + expected: data(12, root1, true, true).Data, + }, + { + name: "ProviderFailureDoesNotVote", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "error": erroringProvider(t, errors.New("failed")), + "valid": provider(t, data(12, root1, true, true)), + }, + threshold: 1, + expected: data(12, root1, true, true).Data, + }, + { + name: "SameRootPayloadPresenceTie", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "absent": provider(t, data(12, root1, false, true)), + "present": provider(t, data(12, root1, true, true)), + }, + threshold: 1, + expected: data(12, root1, true, true).Data, + }, + { + name: "BlobOnlyTieFails", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "unavailable": provider(t, data(12, root1, false, false)), + "available": provider(t, data(12, root1, false, true)), + }, + threshold: 1, + err: "split-response payload attestation data responses", + }, + { + name: "CompetingRootsFail", + providers: map[string]eth2client.PayloadAttestationDataProvider{ + "one": provider(t, data(12, root1, true, true)), + "two": provider(t, data(12, root2, true, true)), + }, + threshold: 1, + err: "split-root payload attestation data responses", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + service, err := majority.New(ctx, + majority.WithLogLevel(zerolog.Disabled), + majority.WithTimeout(time.Second), + majority.WithThreshold(test.threshold), + majority.WithPayloadAttestationDataProviders(test.providers), + ) + require.NoError(t, err) + actual, err := service.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + if test.err != "" { + require.EqualError(t, err, test.err) + return + } + require.NoError(t, err) + require.Equal(t, test.expected, actual.Data) + }) + } +} + +func TestPayloadAttestationDataHonoursCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + service, err := majority.New(ctx, + majority.WithTimeout(time.Second), + majority.WithPayloadAttestationDataProviders(map[string]eth2client.PayloadAttestationDataProvider{ + "slow": erroringProvider(t, context.Canceled), + }), + ) + require.NoError(t, err) + _, err = service.PayloadAttestationData(ctx, &api.PayloadAttestationDataOpts{Slot: 12}) + require.ErrorIs(t, err, context.Canceled) +} diff --git a/util/config.go b/util/config.go index 1df82aa2..8703a830 100644 --- a/util/config.go +++ b/util/config.go @@ -80,7 +80,14 @@ func BeaconNodeAddressesForProposing() []string { // payload attestation data from the configuration. // This follows the hierarchical address configuration, and removes duplicates. func BeaconNodeAddressesForPayloadAttestationData() []string { - return uniqueSortedAddresses(BeaconNodeAddresses("strategies.payloadattestationdata")) + switch viper.GetString("strategies.payloadattestationdata.style") { + case "first": + return uniqueSortedAddresses(BeaconNodeAddresses("strategies.payloadattestationdata.first")) + case "majority": + return uniqueSortedAddresses(BeaconNodeAddresses("strategies.payloadattestationdata.majority")) + default: + return uniqueSortedAddresses(BeaconNodeAddresses("strategies.payloadattestationdata")) + } } // BeaconNodeAddressesForAttestationData obtains the beacon node addresses used for diff --git a/util/config_test.go b/util/config_test.go index 41bf1f6d..b4e3de8b 100644 --- a/util/config_test.go +++ b/util/config_test.go @@ -286,6 +286,29 @@ func TestBeaconNodeAddressesPerStrategy(t *testing.T) { envPrefix: "VOUCH_BEACONNODEADDRESSFORPAYLOADATTESTATIONDATA", handler: util.BeaconNodeAddressesForPayloadAttestationData, }, + { + name: "PayloadAttestationDataFirstStrategyUsesStyleAddresses", + env: map[string]string{ + "BEACON_NODE_ADDRESSES": "1 2", + "STRATEGIES_PAYLOADATTESTATIONDATA_BEACON_NODE_ADDRESSES": "3 4", + "STRATEGIES_PAYLOADATTESTATIONDATA_STYLE": "first", + "STRATEGIES_PAYLOADATTESTATIONDATA_FIRST_BEACON_NODE_ADDRESSES": "5 6", + }, + expected: []string{"5", "6"}, + envPrefix: "VOUCH_BEACONNODEADDRESSFORPAYLOADATTESTATIONDATA", + handler: util.BeaconNodeAddressesForPayloadAttestationData, + }, + { + name: "PayloadAttestationDataMajorityStrategyFallsBackToParentAddresses", + env: map[string]string{ + "BEACON_NODE_ADDRESSES": "1 2", + "STRATEGIES_PAYLOADATTESTATIONDATA_BEACON_NODE_ADDRESSES": "3 4", + "STRATEGIES_PAYLOADATTESTATIONDATA_STYLE": "majority", + }, + expected: []string{"3", "4"}, + envPrefix: "VOUCH_BEACONNODEADDRESSFORPAYLOADATTESTATIONDATA", + handler: util.BeaconNodeAddressesForPayloadAttestationData, + }, } // SignedBeaconBlock and BeaconBlockHeader only support "first" style,