diff --git a/core/aggsigdb/memory.go b/core/aggsigdb/memory.go index c3ae95f83..657997df7 100644 --- a/core/aggsigdb/memory.go +++ b/core/aggsigdb/memory.go @@ -6,17 +6,27 @@ import ( "bytes" "context" + "github.com/attestantio/go-eth2-client/spec/bellatrix" + eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0" + "github.com/obolnetwork/charon/app/errors" "github.com/obolnetwork/charon/core" ) -var ErrStopped = errors.New("database stopped") +var ( + ErrStopped = errors.New("database stopped") + + // errNotAwaitable is returned when awaiting a duty family that is only stored and + // broadcasted, never queried. + errNotAwaitable = errors.New("duty aggregate is not awaitable") +) // NewMemDB creates a basic memory based AggSigDB. func NewMemDB(deadliner core.Deadliner) *MemDB { return &MemDB{ - data: make(map[memDBKey]core.SignedData), - keysByDuty: make(map[core.Duty][]memDBKey), + generalDuties: newAggStore[generalKey](), + subCommDuties: newAggStore[subCommKey](), + propPrefDuties: newAggStore[propPrefKey](), commands: make(chan writeCommand), queries: make(chan readQuery), blockedQueries: []readQuery{}, @@ -28,8 +38,17 @@ func NewMemDB(deadliner core.Deadliner) *MemDB { // MemDB is a basic memory implementation of core.AggSigDB. type MemDB struct { - data map[memDBKey]core.SignedData - keysByDuty map[core.Duty][]memDBKey // Key index by duty for fast deletion. + // Aggregates are stored per duty family, each keyed by the family's identity + // and trimmed when the deadliner expires the duty. + + // generalDuties holds duties with a single message per duty and validator. + generalDuties *aggStore[generalKey] + // subCommDuties holds sync-committee aggregator duties, additionally keyed by + // sync subcommittee index. + subCommDuties *aggStore[subCommKey] + // propPrefDuties holds proposer preferences, additionally keyed by the fields + // that may legitimately change on resubmission. They are not awaitable, only broadcasted. + propPrefDuties *aggStore[propPrefKey] commands chan writeCommand queries chan readQuery @@ -43,12 +62,7 @@ type MemDB struct { // Store implements core.AggSigDB, see its godoc. func (db *MemDB) Store(ctx context.Context, duty core.Duty, set core.SignedDataSet) error { for pubKey, data := range set { - subcommIdx, err := core.SyncSubcommitteeIndex(duty.Type, data) - if err != nil { - return err - } - - if err := db.store(ctx, memDBKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx}, data); err != nil { + if err := db.store(ctx, duty, pubKey, data); err != nil { return err } } @@ -56,7 +70,7 @@ func (db *MemDB) Store(ctx context.Context, duty core.Duty, set core.SignedDataS return nil } -func (db *MemDB) store(ctx context.Context, key memDBKey, data core.SignedData) error { +func (db *MemDB) store(ctx context.Context, duty core.Duty, pubKey core.PubKey, data core.SignedData) error { clone, err := data.Clone() // Clone before storing. if err != nil { return err @@ -64,7 +78,8 @@ func (db *MemDB) store(ctx context.Context, key memDBKey, data core.SignedData) response := make(chan error, 1) cmd := writeCommand{ - memDBKey: key, + duty: duty, + pubKey: pubKey, data: clone, response: response, } @@ -89,15 +104,21 @@ func (db *MemDB) store(ctx context.Context, key memDBKey, data core.SignedData) // Await implements core.AggSigDB, see its godoc. func (db *MemDB) Await(ctx context.Context, duty core.Duty, pubKey core.PubKey, subcommIdx core.SubcommitteeIndex) (core.SignedData, error) { + if duty.Type == core.DutyProposerPreferences { + return nil, errNotAwaitable + } + cancel := make(chan struct{}) defer close(cancel) response := make(chan core.SignedData, 1) query := readQuery{ - memDBKey: memDBKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx}, - response: response, - cancel: cancel, + duty: duty, + pubKey: pubKey, + subcommIdx: subcommIdx, + response: response, + cancel: cancel, } select { @@ -134,11 +155,9 @@ func (db *MemDB) Run(ctx context.Context) { db.callbackBlockedQueriesForT() } case duty := <-db.deadliner.C(): - for _, key := range db.keysByDuty[duty] { - delete(db.data, key) - } - - delete(db.keysByDuty, duty) + db.generalDuties.trim(duty) + db.subCommDuties.trim(duty) + db.propPrefDuties.trim(duty) case <-ctx.Done(): return } @@ -151,39 +170,15 @@ func (db *MemDB) execCommand(command writeCommand) { _ = db.deadliner.Add(command.duty) // TODO(corver): Distinguish between no deadline supported vs already expired. - key := command.memDBKey - - if existing, ok := db.data[key]; ok { - equal, err := dataEqual(existing, command.data) - if err != nil { - command.response <- err - } else if !equal { - command.response <- errors.New("mismatching data") - } - } else { - db.data[key] = command.data - db.keysByDuty[command.duty] = append(db.keysByDuty[command.duty], key) - } -} - -func dataEqual(x core.SignedData, y core.SignedData) (bool, error) { - bx, err := x.MarshalJSON() - if err != nil { - return false, errors.Wrap(err, "marshal data") + if err := db.storeRouted(command.duty, command.pubKey, command.data); err != nil { + command.response <- err } - - by, err := y.MarshalJSON() - if err != nil { - return false, errors.Wrap(err, "marshal data") - } - - return bytes.Equal(bx, by), nil } // execQuery returns true if the query was successfully executed. -// If the requested entry is found in the DB it will return it via query.response channel. +// If the requested entry is found in the DB it will be returned via query.response channel. func (db *MemDB) execQuery(query readQuery) bool { - data, ok := db.data[query.memDBKey] + data, ok := db.get(query.duty, query.pubKey, query.subcommIdx) if !ok { return false } @@ -230,19 +225,10 @@ func cancelled(cancel <-chan struct{}) bool { } } -type memDBKey struct { - duty core.Duty - pubKey core.PubKey - // subcommIdx is the sync subcommittee index for sync-committee aggregator - // duties (DutyPrepareSyncContribution, DutySyncContribution), and 0 otherwise. - // A validator can occupy multiple sync subcommittees in the same slot, so it - // disambiguates their otherwise-colliding aggregated signatures. - subcommIdx core.SubcommitteeIndex -} - // writeCommand holds the data to write into the database. type writeCommand struct { - memDBKey + duty core.Duty + pubKey core.PubKey data core.SignedData response chan<- error @@ -250,8 +236,168 @@ type writeCommand struct { // readQuery holds the query data and the response channel. type readQuery struct { - memDBKey + duty core.Duty + pubKey core.PubKey + subcommIdx core.SubcommitteeIndex response chan<- core.SignedData cancel <-chan struct{} } + +// generalKey identifies aggregates for duties with a single message per duty and validator. +type generalKey struct { + duty core.Duty + pubKey core.PubKey +} + +// subCommKey additionally carries the sync subcommittee index for sync-committee aggregator +// duties (DutyPrepareSyncContribution, DutySyncContribution). A validator can occupy multiple +// sync subcommittees in the same slot, so it disambiguates their otherwise-colliding aggregates. +type subCommKey struct { + duty core.Duty + pubKey core.PubKey + subcommIdx core.SubcommitteeIndex +} + +// propPrefKey additionally carries the proposer preferences fields that may legitimately change +// for the same duty and pubkey: a reorg changes the dependent root, or operators change the fee +// recipient or gas limit in sync, and a new aggregate reaches threshold. It disambiguates the +// aggregates so each message is stored independently. +type propPrefKey struct { + duty core.Duty + pubKey core.PubKey + dependentRoot eth2p0.Root + feeRecipient bellatrix.ExecutionAddress + targetGasLimit uint64 +} + +// propPrefKeyFor returns the propPrefKey for the provided proposer preferences aggregate. +func propPrefKeyFor(duty core.Duty, pubKey core.PubKey, data core.SignedData) (propPrefKey, error) { + pref, ok := data.(core.SignedProposerPreferences) + if !ok || pref.Message == nil { + return propPrefKey{}, errors.New("invalid proposer preferences data") + } + + return propPrefKey{ + duty: duty, + pubKey: pubKey, + dependentRoot: pref.Message.DependentRoot, + feeRecipient: pref.Message.FeeRecipient, + targetGasLimit: pref.Message.TargetGasLimit, + }, nil +} + +// storeRouted routes the aggregate to its duty family store. Callers must clone data +// before storing. +func (db *MemDB) storeRouted(duty core.Duty, pubKey core.PubKey, data core.SignedData) error { + switch duty.Type { + case core.DutyPrepareSyncContribution, core.DutySyncContribution: + subcommIdx, err := core.SyncSubcommitteeIndex(duty.Type, data) + if err != nil { + return err + } + + return db.subCommDuties.store(duty, subCommKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx}, data) + case core.DutyProposerPreferences: + k, err := propPrefKeyFor(duty, pubKey, data) + if err != nil { + return err + } + + return db.propPrefDuties.store(duty, k, data) + default: + return db.generalDuties.store(duty, generalKey{duty: duty, pubKey: pubKey}, data) + } +} + +// get returns the aggregate for the provided awaitable duty. Proposer preferences are not +// awaitable, Await rejects them before querying. +func (db *MemDB) get(duty core.Duty, pubKey core.PubKey, subcommIdx core.SubcommitteeIndex) (core.SignedData, bool) { + switch duty.Type { + case core.DutyPrepareSyncContribution, core.DutySyncContribution: + return db.subCommDuties.get(subCommKey{duty: duty, pubKey: pubKey, subcommIdx: subcommIdx}) + default: + return db.generalDuties.get(generalKey{duty: duty, pubKey: pubKey}) + } +} + +// newAggStore returns a new empty aggregate store. +func newAggStore[K comparable]() *aggStore[K] { + return &aggStore[K]{ + data: make(map[K]core.SignedData), + keysByDuty: make(map[core.Duty][]K), + } +} + +// aggStore holds aggregates for one duty family, keyed by the family's identity. +// It is not thread safe, synchronisation is up to the caller. +type aggStore[K comparable] struct { + data map[K]core.SignedData + keysByDuty map[core.Duty][]K // Key index by duty for fast deletion. +} + +// store stores the aggregate at the provided key. Storing an identical aggregate again is a +// no-op, storing a different aggregate at an existing key is an error. +func (s *aggStore[K]) store(duty core.Duty, k K, data core.SignedData) error { + if existing, ok := s.data[k]; ok { + equal, err := dataEqual(existing, data) + if err != nil { + return err + } else if !equal { + return errors.New("mismatching data") + } + + return nil + } + + s.data[k] = data + s.keysByDuty[duty] = append(s.keysByDuty[duty], k) + + return nil +} + +// get returns the aggregate stored at the provided key. +func (s *aggStore[K]) get(k K) (core.SignedData, bool) { + data, ok := s.data[k] + + return data, ok +} + +// trim deletes all aggregates for the provided duty. It writes nothing for unknown duties. +func (s *aggStore[K]) trim(duty core.Duty) { + keys, ok := s.keysByDuty[duty] + if !ok { + return + } + + for _, k := range keys { + delete(s.data, k) + } + + delete(s.keysByDuty, duty) +} + +// dataEqual returns true if the provided signed data is equal. +func dataEqual(x core.SignedData, y core.SignedData) (bool, error) { + bx, err := x.MarshalJSON() + if err != nil { + return false, errors.Wrap(err, "marshal data") + } + + by, err := y.MarshalJSON() + if err != nil { + return false, errors.Wrap(err, "marshal data") + } + + return bytes.Equal(bx, by), nil +} + +// memDBKey identifies aggregates in the legacy MemDBV2 implementation. +// TODO(kalo): remove together with MemDBV2, it is unused by MemDB. +type memDBKey struct { + duty core.Duty + pubKey core.PubKey + // subcommIdx is the sync subcommittee index for sync-committee aggregator + // duties (DutyPrepareSyncContribution, DutySyncContribution), and 0 otherwise. + subcommIdx core.SubcommitteeIndex +} diff --git a/core/aggsigdb/memory_internal_test.go b/core/aggsigdb/memory_internal_test.go index 6d71db92e..e43f8456f 100644 --- a/core/aggsigdb/memory_internal_test.go +++ b/core/aggsigdb/memory_internal_test.go @@ -47,8 +47,8 @@ func TestDutyExpiration(t *testing.T) { cancel() wg.Wait() - require.Empty(t, db.data) - require.Empty(t, db.keysByDuty) + require.Empty(t, db.generalDuties.data) + require.Empty(t, db.generalDuties.keysByDuty) } func TestCancelledQuery(t *testing.T) { @@ -146,3 +146,39 @@ func (d *testDeadliner) Expire() { d.added = nil } + +// TestProposerPreferencesReorg verifies that aggregates for the same proposer preferences duty +// and pubkey but different message roots (a reorg changed the dependent root and a new aggregate +// reached threshold) are stored independently instead of erroring as mismatching data. +func TestProposerPreferencesReorg(t *testing.T) { + var wg sync.WaitGroup + + ctx, cancel := context.WithCancel(context.Background()) + + db := NewMemDB(newTestDeadliner()) + + wg.Go(func() { + db.Run(ctx) + }) + + pubkey := testutil.RandomCorePubKey(t) + + prefA := testutil.RandomProposerPreferences() + duty := core.NewProposerPreferencesDuty(uint64(prefA.Message.ProposalSlot)) + + // Same slot and validator, different dependent root (reorg). + prefB := testutil.RandomProposerPreferences() + prefB.Message.ProposalSlot = prefA.Message.ProposalSlot + prefB.Message.ValidatorIndex = prefA.Message.ValidatorIndex + + err := db.Store(ctx, duty, core.SignedDataSet{pubkey: core.NewSignedProposerPreferences(prefA)}) + require.NoError(t, err) + + err = db.Store(ctx, duty, core.SignedDataSet{pubkey: core.NewSignedProposerPreferences(prefB)}) + require.NoError(t, err) + + cancel() + wg.Wait() + + require.Len(t, db.propPrefDuties.data, 2) +} diff --git a/core/bcast/bcast.go b/core/bcast/bcast.go index ae23cd2aa..5979ef328 100644 --- a/core/bcast/bcast.go +++ b/core/bcast/bcast.go @@ -248,6 +248,13 @@ func (b Broadcaster) Broadcast(ctx context.Context, duty core.Duty, set core.Sig return nil case core.DutyPrepareAggregator: // Beacon committee selections are only applicable to DVT, not broadcasted to beacon chain. + return nil + case core.DutyProposerPreferences: + // TODO(gloas): submit the aggregated SignedProposerPreferences to the beacon node once + // go-eth2-client supports it (attestantio/go-eth2-client#316). No-op meanwhile so + // reaching threshold doesn't fail the intake path. + log.Debug(ctx, "Proposer preferences submission not yet supported, skipping broadcast") + return nil case core.DutyAggregator: aggAndProofs, err := setToAggAndProof(set) diff --git a/core/bcast/bcast_test.go b/core/bcast/bcast_test.go index 91f90605f..95452d4bc 100644 --- a/core/bcast/bcast_test.go +++ b/core/bcast/bcast_test.go @@ -91,6 +91,10 @@ func TestBroadcastOtherDuties(t *testing.T) { err = bcaster.Broadcast(context.Background(), core.Duty{Type: core.DutyPrepareSyncContribution}, nil) require.NoError(t, err) + // TODO(gloas): replace with a real submission test once go-eth2-client supports it. + err = bcaster.Broadcast(context.Background(), core.Duty{Type: core.DutyProposerPreferences}, nil) + require.NoError(t, err) + err = bcaster.Broadcast(context.Background(), core.Duty{Type: core.DutyUnknown}, nil) require.ErrorContains(t, err, "unsupported duty type") } diff --git a/core/deadline.go b/core/deadline.go index 9fa2e48ba..6dcaaa04c 100644 --- a/core/deadline.go +++ b/core/deadline.go @@ -135,6 +135,11 @@ func NewDutyDeadlineFunc(ctx context.Context, eth2Cl eth2wrap.Client) (DeadlineF case DutyPayloadAttestation: // Payload attestation messages are only accepted on gossip within their own slot. duration = slotDuration + case DutyProposerPreferences: + // Preferences are submitted from the start of the epoch before the proposal epoch + // (the earliest the assignment is derivable) and are keyed by the proposal slot, + // so entries live from submission (up to ~2 epochs early) until the slot passes. + duration = slotDuration default: duration = slotDuration } diff --git a/core/eth2signeddata.go b/core/eth2signeddata.go index 05939075c..d2f4f916d 100644 --- a/core/eth2signeddata.go +++ b/core/eth2signeddata.go @@ -27,6 +27,7 @@ var ( _ Eth2SignedData = SignedSyncContributionAndProof{} _ Eth2SignedData = SyncCommitteeSelection{} _ Eth2SignedData = VersionedPayloadAttestationMessage{} + _ Eth2SignedData = SignedProposerPreferences{} ) // VerifyEth2SignedData verifies signature associated with the given Eth2SignedData. @@ -186,3 +187,17 @@ func (m VersionedPayloadAttestationMessage) Epoch(ctx context.Context, eth2Cl et return eth2util.EpochFromSlot(ctx, eth2Cl, data.Slot) } + +// Implement Eth2SignedData for SignedProposerPreferences. + +func (SignedProposerPreferences) DomainName() signing.DomainName { + return signing.DomainProposerPreferences +} + +func (p SignedProposerPreferences) Epoch(ctx context.Context, eth2Cl eth2wrap.Client) (eth2p0.Epoch, error) { + if p.Message == nil { + return 0, errors.New("nil proposer preferences message") + } + + return eth2util.EpochFromSlot(ctx, eth2Cl, p.Message.ProposalSlot) +} diff --git a/core/eth2signeddata_test.go b/core/eth2signeddata_test.go index 4a423d694..d2ef6698d 100644 --- a/core/eth2signeddata_test.go +++ b/core/eth2signeddata_test.go @@ -82,6 +82,10 @@ func TestVerifyEth2SignedData(t *testing.T) { name: "verify sync contribution and proof", data: testutil.RandomCoreSignedSyncContributionAndProof(), }, + { + name: "verify proposer preferences", + data: core.NewSignedProposerPreferences(testutil.RandomProposerPreferences()), + }, } for _, test := range tests { diff --git a/core/interfaces.go b/core/interfaces.go index 55446da62..a5c5faff5 100644 --- a/core/interfaces.go +++ b/core/interfaces.go @@ -209,6 +209,11 @@ type AggSigDB interface { // (DutyPrepareSyncContribution, DutySyncContribution), where a validator can have // a distinct aggregated signature per subcommittee in a slot; it is 0 for all // other duties (see IsSyncSubcommitteeDuty). + // + // TODO(kalo): Refactor the query API so per-duty-family fields aren't forced on every + // caller: subcommIdx only applies to the sync-committee aggregator duties, all other + // callers pass a meaningless 0, and duty families whose identity exceeds these + // parameters (proposer preferences) cannot be addressed at all. Await(context context.Context, duty Duty, pubKey PubKey, subcommIdx SubcommitteeIndex) (SignedData, error) // Run runs AggSigDB lifecycle until context is cancelled. diff --git a/core/parsigdb/memory.go b/core/parsigdb/memory.go index 45370f226..0224f602d 100644 --- a/core/parsigdb/memory.go +++ b/core/parsigdb/memory.go @@ -10,6 +10,9 @@ import ( "sync" "time" + "github.com/attestantio/go-eth2-client/spec/bellatrix" + eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0" + "github.com/obolnetwork/charon/app/errors" "github.com/obolnetwork/charon/app/log" "github.com/obolnetwork/charon/app/z" @@ -40,12 +43,14 @@ type MemDBMetadata struct { // NewMemDB returns a new in-memory partial signature database instance. func NewMemDB(threshold int, deadliner core.Deadliner, metadata MemDBMetadata) *MemDB { return &MemDB{ - entries: make(map[key][]core.ParSignedData), - keysByDuty: make(map[core.Duty][]key), - exemptEntries: make(map[exemptEntryKey][]key), - threshold: threshold, - deadliner: deadliner, - metadata: metadata, + generalDuties: newDutyStore[generalKey](), + subCommDuties: newDutyStore[subCommKey](), + propPrefDuties: newDutyStore[propPrefKey](), + persistedDuties: newDutyStore[generalKey](), + exemptEntries: make(map[exemptEntryKey][]generalKey), + threshold: threshold, + deadliner: deadliner, + metadata: metadata, } } @@ -55,14 +60,26 @@ type MemDB struct { internalSubs []func(context.Context, core.Duty, core.ParSignedDataSet) error threshSubs []func(context.Context, core.Duty, map[core.PubKey][]core.ParSignedData) error - entries map[key][]core.ParSignedData - keysByDuty map[core.Duty][]key - // exemptEntries indexes exempt-duty entries (which the deadliner never trims) by - // (share index, validator, duty type) in insertion order, so they can be capped and - // evicted oldest-first to bound memory. - exemptEntries map[exemptEntryKey][]key - threshold int - deadliner core.Deadliner + // Partial signatures are stored per duty family, each keyed by the family's + // aggregation identity and trimmed when the deadliner expires the duty. + + // generalDuties holds duties with a single message per duty and validator. + generalDuties *dutyStore[generalKey] + // subCommDuties holds sync-committee aggregator duties, additionally keyed by + // sync subcommittee index. + subCommDuties *dutyStore[subCommKey] + // propPrefDuties holds proposer preferences, additionally keyed by the fields + // that may legitimately change on resubmission. + propPrefDuties *dutyStore[propPrefKey] + // persistedDuties holds exempt duties (exits, builder registrations) which the + // deadliner never trims; they are capped via exemptEntries instead. + persistedDuties *dutyStore[generalKey] + // exemptEntries indexes persisted-duty entries by (share index, validator, duty type) + // in insertion order, so they can be capped and evicted oldest-first to bound memory. + exemptEntries map[exemptEntryKey][]generalKey + + threshold int + deadliner core.Deadliner metadata MemDBMetadata } @@ -134,12 +151,7 @@ func (db *MemDB) StoreExternal(ctx context.Context, duty core.Duty, signedSet co output := make(map[core.PubKey][]core.ParSignedData) for pubkey, sig := range signedSet { - subcommIdx, err := core.SyncSubcommitteeIndex(duty.Type, sig.SignedData) - if err != nil { - return err - } - - sigs, ok, err := db.store(ctx, key{Duty: duty, PubKey: pubkey, SubcommIdx: subcommIdx}, sig, exempt) + sigs, ok, err := db.store(ctx, duty, pubkey, sig, exempt) if err != nil { return err } else if !ok { @@ -183,29 +195,27 @@ func (db *MemDB) Trim(ctx context.Context) { return case duty := <-db.deadliner.C(): // This buffered channel is small, so we need dedicated goroutine to service it. db.mu.Lock() - - for _, key := range db.keysByDuty[duty] { - delete(db.entries, key) - } - - delete(db.keysByDuty, duty) + // Persisted duties are never emitted on deadliner.C(), so they are not trimmed. + db.generalDuties.trim(duty) + db.subCommDuties.trim(duty) + db.propPrefDuties.trim(duty) db.mu.Unlock() } } } -// store returns true if the value was added to the list of signatures at the provided key -// and returns a copy of the resulting list. -func (db *MemDB) store(ctx context.Context, k key, value core.ParSignedData, exempt bool) ([]core.ParSignedData, bool, error) { +// store routes the value to its duty family store and returns true and a copy of the +// resulting signature list if it was added. +func (db *MemDB) store(ctx context.Context, duty core.Duty, pubkey core.PubKey, value core.ParSignedData, exempt bool) ([]core.ParSignedData, bool, error) { db.mu.Lock() defer db.mu.Unlock() now := time.Now().UnixMilli() - slotStart := (uint64(db.metadata.genesisTime.Unix()) + k.Duty.Slot*db.metadata.slotDuration) * 1000 // in ms - timeSinceSlotStart := float64(now-int64(slotStart)) / 1000 // in seconds + slotStart := (uint64(db.metadata.genesisTime.Unix()) + duty.Slot*db.metadata.slotDuration) * 1000 // in ms + timeSinceSlotStart := float64(now-int64(slotStart)) / 1000 // in seconds - switch k.Duty.Type { + switch duty.Type { case core.DutyAttester: timeSinceSlotStart -= 4.0 case core.DutyAggregator, core.DutySyncContribution: @@ -215,47 +225,61 @@ func (db *MemDB) store(ctx context.Context, k key, value core.ParSignedData, exe // Observe time since slot start for received partial signatures, with share index as label for better visibility of late partial signatures. // Subtracting 1 from share index to have 0-based index. - parsigStored.WithLabelValues(k.Duty.Type.String(), strconv.FormatInt(int64(value.ShareIdx-1), 10)).Observe(timeSinceSlotStart) + parsigStored.WithLabelValues(duty.Type.String(), strconv.FormatInt(int64(value.ShareIdx-1), 10)).Observe(timeSinceSlotStart) - for _, s := range db.entries[k] { - if s.ShareIdx == value.ShareIdx { - equal, err := parSignedDataEqual(s, value) - if err != nil { - return nil, false, err - } else if !equal { - return nil, false, errors.New("mismatching partial signed data", - z.Any("pubkey", k.PubKey), z.Int("share_idx", s.ShareIdx)) - } + var ( + sigs []core.ParSignedData + added bool + err error + ) - return nil, false, nil + switch { + case exempt: + // Persisted duties are never emitted on deadliner.C(), so they are tracked and + // capped here (under the same lock) instead of being indexed for trimming. + k := generalKey{Duty: duty, PubKey: pubkey} + + sigs, added, err = db.persistedDuties.store(duty, pubkey, k, value, false) + if err == nil && added { + db.trackExemptUnsafe(ctx, k, value.ShareIdx) } - } + case duty.Type == core.DutyPrepareSyncContribution || duty.Type == core.DutySyncContribution: + var subcommIdx core.SubcommitteeIndex - // Clone before storing. - clone, err := value.Clone() - if err != nil { - return nil, false, err - } + subcommIdx, err = core.SyncSubcommitteeIndex(duty.Type, value.SignedData) + if err != nil { + return nil, false, err + } - isNewKey := len(db.entries[k]) == 0 + sigs, added, err = db.subCommDuties.store(duty, pubkey, subCommKey{Duty: duty, PubKey: pubkey, SubcommIdx: subcommIdx}, value, true) + case duty.Type == core.DutyProposerPreferences: + pref, ok := value.SignedData.(core.SignedProposerPreferences) + if !ok || pref.Message == nil { + return nil, false, errors.New("invalid proposer preferences data") + } - db.entries[k] = append(db.entries[k], clone) + k := propPrefKey{ + Duty: duty, + PubKey: pubkey, + DependentRoot: pref.Message.DependentRoot, + FeeRecipient: pref.Message.FeeRecipient, + TargetGasLimit: pref.Message.TargetGasLimit, + } - if exempt { - // Exempt duties are never emitted on deadliner.C(), so they are tracked and capped - // here (under the same lock) instead of being trimmed via keysByDuty. - db.trackExemptUnsafe(ctx, k, value.ShareIdx) - } else if isNewKey { - // Index each key once; Trim deletes by key, so appending per share signature would - // only add redundant duplicates (O(validators*shares) instead of O(validators)). - db.keysByDuty[k.Duty] = append(db.keysByDuty[k.Duty], k) + sigs, added, err = db.propPrefDuties.store(duty, pubkey, k, value, true) + default: + sigs, added, err = db.generalDuties.store(duty, pubkey, generalKey{Duty: duty, PubKey: pubkey}, value, true) + } + + if err != nil { + return nil, false, err } - if k.Duty.Type == core.DutyExit { - exitCounter.WithLabelValues(k.PubKey.String()).Inc() + if added && duty.Type == core.DutyExit { + exitCounter.WithLabelValues(pubkey.String()).Inc() } - return append([]core.ParSignedData(nil), db.entries[k]...), true, nil + return sigs, added, nil } // clone returns a deep copy of the provided map. @@ -326,16 +350,98 @@ func parSignedDataEqual(x, y core.ParSignedData) (bool, error) { return bytes.Equal(xjson, yjson), nil } -type key struct { +// generalKey identifies partial signatures for duties with a single message per duty +// and validator. +type generalKey struct { Duty core.Duty PubKey core.PubKey - // SubcommIdx is the sync subcommittee index for sync-committee aggregator - // duties (DutyPrepareSyncContribution, DutySyncContribution), and 0 otherwise. - // A validator can occupy multiple sync subcommittees in the same slot, so it - // disambiguates their otherwise-colliding partial signatures. +} + +// subCommKey additionally carries the sync subcommittee index for sync-committee aggregator +// duties (DutyPrepareSyncContribution, DutySyncContribution). A validator can occupy multiple +// sync subcommittees in the same slot, so it disambiguates their otherwise-colliding partial +// signatures. +type subCommKey struct { + Duty core.Duty + PubKey core.PubKey SubcommIdx core.SubcommitteeIndex } +// propPrefKey additionally carries the proposer preferences fields that may legitimately change +// for the same duty and pubkey: a reorg changes the dependent root, or operators change the fee +// recipient or gas limit in sync, and VCs resubmit. It disambiguates the resubmission from the +// original partial signatures so each message aggregates independently, and unlike an opaque +// message root it shows what differs between coexisting entries. +type propPrefKey struct { + Duty core.Duty + PubKey core.PubKey + DependentRoot eth2p0.Root + FeeRecipient bellatrix.ExecutionAddress + TargetGasLimit uint64 +} + +// newDutyStore returns a new empty duty store. +func newDutyStore[K comparable]() *dutyStore[K] { + return &dutyStore[K]{ + entries: make(map[K][]core.ParSignedData), + keysByDuty: make(map[core.Duty][]K), + } +} + +// dutyStore holds partial signatures for one duty family, keyed by the family's +// aggregation identity. It is not thread safe, callers must hold the MemDB lock. +type dutyStore[K comparable] struct { + entries map[K][]core.ParSignedData + keysByDuty map[core.Duty][]K +} + +// store stores the value at the provided key and returns a copy of the resulting signature +// list and true. It returns false if the share already stored an identical value (duplicate) +// and an error if the share already stored a different value. If index is false, the key is +// not indexed for trimming (persisted duties are capped by the caller instead). +func (s *dutyStore[K]) store(duty core.Duty, pubkey core.PubKey, k K, value core.ParSignedData, index bool) ([]core.ParSignedData, bool, error) { + for _, existing := range s.entries[k] { + if existing.ShareIdx == value.ShareIdx { + equal, err := parSignedDataEqual(existing, value) + if err != nil { + return nil, false, err + } else if !equal { + return nil, false, errors.New("mismatching partial signed data", + z.Any("duty", duty), z.Any("pubkey", pubkey), z.Int("share_idx", value.ShareIdx)) + } + + return nil, false, nil + } + } + + // Clone before storing. + clone, err := value.Clone() + if err != nil { + return nil, false, err + } + + isNewKey := len(s.entries[k]) == 0 + + s.entries[k] = append(s.entries[k], clone) + + if index && isNewKey { + // Index each key once; trim deletes by key, so appending per share signature would + // only add redundant duplicates (O(validators*shares) instead of O(validators)). + s.keysByDuty[duty] = append(s.keysByDuty[duty], k) + } + + return append([]core.ParSignedData(nil), s.entries[k]...), true, nil +} + +// trim deletes all entries for the provided duty. +func (s *dutyStore[K]) trim(duty core.Duty) { + for _, k := range s.keysByDuty[duty] { + delete(s.entries, k) + } + + delete(s.keysByDuty, duty) +} + // exemptEntryKey indexes exempt-duty entries by share index, validator and duty type. // Distinct entries within this key correspond to distinct duties (e.g. exit epochs). type exemptEntryKey struct { @@ -346,7 +452,7 @@ type exemptEntryKey struct { // trackExemptUnsafe records a newly stored exempt-duty entry for capping. It assumes db.mu is held // and that k was just added as a new entry for shareIdx. -func (db *MemDB) trackExemptUnsafe(ctx context.Context, k key, shareIdx int) { +func (db *MemDB) trackExemptUnsafe(ctx context.Context, k generalKey, shareIdx int) { ek := exemptEntryKey{ShareIdx: shareIdx, PubKey: k.PubKey, DutyType: k.Duty.Type} stored := db.exemptEntries[ek] @@ -375,7 +481,7 @@ func (db *MemDB) trackExemptUnsafe(ctx context.Context, k key, shareIdx int) { // evictExemptShareEntryUnsafe removes the given share's partial signature from the entry at k, // deleting the entry entirely if no other shares remain. It warns with the evicted data for // forensics, since eviction only happens when a share exceeds the per-share cap. It assumes db.mu is held. -func (db *MemDB) evictExemptShareEntryUnsafe(ctx context.Context, k key, shareIdx int) { +func (db *MemDB) evictExemptShareEntryUnsafe(ctx context.Context, k generalKey, shareIdx int) { // Log the evicted key (not the data) for forensics; this is a hot path, so avoid marshaling. log.Warn(ctx, "Evicting oldest exempt partial signature exceeding per-share cap", nil, z.Any("duty", k.Duty), @@ -384,7 +490,7 @@ func (db *MemDB) evictExemptShareEntryUnsafe(ctx context.Context, k key, shareId z.Int("max_allowed_sigs_per_share", maxExemptEntriesPerShare), ) - sigs := db.entries[k] + sigs := db.persistedDuties.entries[k] remaining := sigs[:0] for _, sig := range sigs { @@ -394,8 +500,8 @@ func (db *MemDB) evictExemptShareEntryUnsafe(ctx context.Context, k key, shareId } if len(remaining) == 0 { - delete(db.entries, k) + delete(db.persistedDuties.entries, k) } else { - db.entries[k] = remaining + db.persistedDuties.entries[k] = remaining } } diff --git a/core/parsigdb/memory_internal_test.go b/core/parsigdb/memory_internal_test.go index 3dbe0c0cd..491677f3a 100644 --- a/core/parsigdb/memory_internal_test.go +++ b/core/parsigdb/memory_internal_test.go @@ -9,6 +9,7 @@ import ( eth2v1 "github.com/attestantio/go-eth2-client/api/v1" "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" "github.com/stretchr/testify/require" @@ -182,7 +183,8 @@ func TestMemDBStoreExternalExpired(t *testing.T) { require.NoError(t, err) db.mu.Lock() - gotEntries, gotKeys := len(db.entries), len(db.keysByDuty) + gotEntries := len(db.generalDuties.entries) + len(db.persistedDuties.entries) + gotKeys := len(db.generalDuties.keysByDuty) db.mu.Unlock() if tt.wantStored { @@ -225,7 +227,7 @@ func TestMemDBExemptCap(t *testing.T) { db.mu.Lock() defer db.mu.Unlock() - require.Len(t, db.entries, maxExemptEntriesPerShare, "entries must be capped at maxExemptEntriesPerShare") + require.Len(t, db.persistedDuties.entries, maxExemptEntriesPerShare, "entries must be capped at maxExemptEntriesPerShare") require.Len(t, db.exemptEntries[exemptEntryKey{ShareIdx: shareIdx, PubKey: pubkey, DutyType: core.DutyAttester}], maxExemptEntriesPerShare) } @@ -269,3 +271,65 @@ func (t *testDeadliner) Add(duty core.Duty) core.DeadlineStatus { func (t *testDeadliner) C() <-chan core.Duty { return t.ch } + +// TestMemDBProposerPreferencesReorg verifies that a share resubmitting proposer preferences for the +// same proposal slot with a different dependent root (after a reorg) does not collide with its +// original partial signature, and that each message root aggregates to threshold independently. +func TestMemDBProposerPreferencesReorg(t *testing.T) { + const th = 2 + + db := NewMemDB(th, newTestDeadliner(), NewMemDBMetadata(eth2util.Mainnet.SlotDuration, time.Unix(eth2util.Mainnet.GenesisTimestamp, 0))) + + var thresholdSets []map[core.PubKey][]core.ParSignedData + + db.SubscribeThreshold(func(_ context.Context, _ core.Duty, set map[core.PubKey][]core.ParSignedData) error { + thresholdSets = append(thresholdSets, set) + + return nil + }) + + pubkey := testutil.RandomCorePubKey(t) + + prefA := testutil.RandomProposerPreferences() + duty := core.NewProposerPreferencesDuty(uint64(prefA.Message.ProposalSlot)) + + // Same slot and validator, different dependent root (reorg). + prefB := testutil.RandomProposerPreferences() + prefB.Message.ProposalSlot = prefA.Message.ProposalSlot + prefB.Message.ValidatorIndex = prefA.Message.ValidatorIndex + + store := func(pref *gloas.SignedProposerPreferences, shareIdx int) error { + return db.StoreExternal(context.Background(), duty, core.ParSignedDataSet{ + pubkey: core.NewPartialSignedProposerPreferences(pref, shareIdx), + }) + } + + // Share 1 submits preferences with dependent root A. + require.NoError(t, store(prefA, 1)) + require.Empty(t, thresholdSets) + + // Share 1 resubmits with dependent root B after a reorg: must not error as mismatching. + require.NoError(t, store(prefB, 1)) + require.Empty(t, thresholdSets) + + // Share 1 resubmits with the same dependent root but a changed gas limit (operators + // changing preferences in sync): must not error as mismatching either. + msgC := *prefB.Message + msgC.TargetGasLimit++ + prefC := *prefB + prefC.Message = &msgC + require.NoError(t, store(&prefC, 1)) + require.Empty(t, thresholdSets) + + // Share 2 submits dependent root B: threshold reached for B only. + require.NoError(t, store(prefB, 2)) + require.Len(t, thresholdSets, 1) + require.Len(t, thresholdSets[0][pubkey], th) + + root, err := thresholdSets[0][pubkey][0].MessageRoot() + require.NoError(t, err) + + rootB, err := core.NewSignedProposerPreferences(prefB).MessageRoot() + require.NoError(t, err) + require.Equal(t, rootB, root) +} diff --git a/core/proto.go b/core/proto.go index fb0342cf8..95b449577 100644 --- a/core/proto.go +++ b/core/proto.go @@ -155,6 +155,13 @@ func ParSignedDataFromProto(typ DutyType, data *pbv1.ParSignedData) (_ ParSigned return ParSignedData{}, errors.Wrap(err, "unmarshal payload attestation message") } + signedData = s + case DutyProposerPreferences: + var s SignedProposerPreferences + if err := unmarshal(data.GetData(), &s); err != nil { + return ParSignedData{}, errors.Wrap(err, "unmarshal signed proposer preferences") + } + signedData = s default: return ParSignedData{}, errors.New("unsupported duty type") diff --git a/core/proto_test.go b/core/proto_test.go index c1a81745a..894ab2149 100644 --- a/core/proto_test.go +++ b/core/proto_test.go @@ -100,6 +100,10 @@ func TestParSignedDataSetProto(t *testing.T) { Type: core.DutySyncContribution, Data: core.NewSignedSyncContributionAndProof(testutil.RandomSignedSyncContributionAndProof()), }, + { + Type: core.DutyProposerPreferences, + Data: core.NewSignedProposerPreferences(testutil.RandomProposerPreferences()), + }, } for _, test := range tests { t.Run(test.Type.String(), func(t *testing.T) { diff --git a/core/serialise_test.go b/core/serialise_test.go index b68370148..6ae8ab3d0 100644 --- a/core/serialise_test.go +++ b/core/serialise_test.go @@ -38,6 +38,7 @@ var coreTypeFuncs = []func() any{ func() any { return new(core.SyncContribution) }, func() any { return new(core.VersionedPayloadAttestationData) }, func() any { return new(core.VersionedPayloadAttestationMessage) }, + func() any { return new(core.SignedProposerPreferences) }, } //go:generate go test . -run=TestJSONSerialisation -update diff --git a/core/signeddata.go b/core/signeddata.go index e06960f2e..6c96634c7 100644 --- a/core/signeddata.go +++ b/core/signeddata.go @@ -50,6 +50,7 @@ var ( _ SignedData = SignedSyncContributionAndProof{} _ SignedData = SyncCommitteeSelection{} _ SignedData = VersionedPayloadAttestationMessage{} + _ SignedData = SignedProposerPreferences{} // Some types support SSZ marshalling and unmarshalling. _ sszMarshaler = VersionedSignedProposal{} @@ -60,6 +61,7 @@ var ( _ sszMarshaler = SyncContributionAndProof{} _ sszMarshaler = SignedSyncContributionAndProof{} _ sszMarshaler = VersionedPayloadAttestationMessage{} + _ sszMarshaler = SignedProposerPreferences{} _ sszUnmarshaler = new(VersionedSignedProposal) _ sszUnmarshaler = new(VersionedAttestation) _ sszUnmarshaler = new(SignedAggregateAndProof) @@ -68,6 +70,7 @@ var ( _ sszUnmarshaler = new(SyncContributionAndProof) _ sszUnmarshaler = new(SignedSyncContributionAndProof) _ sszUnmarshaler = new(VersionedPayloadAttestationMessage) + _ sszUnmarshaler = new(SignedProposerPreferences) ) // SigFromETH2 returns a new signature from eth2 phase0 BLSSignature. @@ -2293,3 +2296,89 @@ type versionedRawPayloadAttMsgJSON struct { Version eth2util.DataVersion `json:"version"` Message json.RawMessage `json:"message"` } + +// ProposerPreferences: https://github.com/ethereum/consensus-specs/blob/master/specs/gloas/validator.md#proposer-preferences. +// From the gloas fork, signed proposer preferences supersede prepare_beacon_proposer and +// register_validator as the source of a proposer's fee recipient and gas limit preferences. + +// NewSignedProposerPreferences is a convenience function which returns a new signed SignedProposerPreferences. +func NewSignedProposerPreferences(data *gloas.SignedProposerPreferences) SignedProposerPreferences { + return SignedProposerPreferences{SignedProposerPreferences: *data} +} + +// NewPartialSignedProposerPreferences is a convenience function which returns a new partially signed SignedProposerPreferences. +func NewPartialSignedProposerPreferences(data *gloas.SignedProposerPreferences, shareIdx int) ParSignedData { + return ParSignedData{ + SignedData: NewSignedProposerPreferences(data), + ShareIdx: shareIdx, + } +} + +// SignedProposerPreferences wraps gloas.SignedProposerPreferences and implements SignedData. +type SignedProposerPreferences struct { + gloas.SignedProposerPreferences +} + +// MessageRoot returns the hash tree root of the proposer preferences message, which is the object +// signed over with DOMAIN_PROPOSER_PREFERENCES at the proposal epoch. +func (p SignedProposerPreferences) MessageRoot() ([32]byte, error) { + if p.Message == nil { + return [32]byte{}, errors.New("nil proposer preferences message") + } + + return p.Message.HashTreeRoot() +} + +func (p SignedProposerPreferences) Signature() Signature { + return SigFromETH2(p.SignedProposerPreferences.Signature) +} + +func (p SignedProposerPreferences) SetSignature(sig Signature) (SignedData, error) { + resp, err := p.clone() + if err != nil { + return nil, err + } + + resp.SignedProposerPreferences.Signature = sig.ToETH2() + + return resp, nil +} + +func (p SignedProposerPreferences) Clone() (SignedData, error) { + return p.clone() +} + +func (p SignedProposerPreferences) clone() (SignedProposerPreferences, error) { + var resp SignedProposerPreferences + + err := cloneSSZMarshaler(p, &resp) + if err != nil { + return SignedProposerPreferences{}, errors.Wrap(err, "clone signed proposer preferences") + } + + return resp, nil +} + +func (p SignedProposerPreferences) MarshalJSON() ([]byte, error) { + return p.SignedProposerPreferences.MarshalJSON() +} + +func (p *SignedProposerPreferences) UnmarshalJSON(input []byte) error { + return p.SignedProposerPreferences.UnmarshalJSON(input) +} + +func (p SignedProposerPreferences) MarshalSSZ() ([]byte, error) { + return p.SignedProposerPreferences.MarshalSSZ() +} + +func (p SignedProposerPreferences) MarshalSSZTo(dst []byte) ([]byte, error) { + return p.SignedProposerPreferences.MarshalSSZTo(dst) +} + +func (p SignedProposerPreferences) SizeSSZ() int { + return p.SignedProposerPreferences.SizeSSZ() +} + +func (p *SignedProposerPreferences) UnmarshalSSZ(b []byte) error { + return p.SignedProposerPreferences.UnmarshalSSZ(b) +} diff --git a/core/signeddata_test.go b/core/signeddata_test.go index 5dbfd6ba8..14ad268db 100644 --- a/core/signeddata_test.go +++ b/core/signeddata_test.go @@ -457,6 +457,10 @@ func TestSignedDataSetSignature(t *testing.T) { name: "signed sync committee selection", data: core.NewSyncCommitteeSelection(testutil.RandomSyncCommitteeSelection()), }, + { + name: "signed proposer preferences", + data: core.NewSignedProposerPreferences(testutil.RandomProposerPreferences()), + }, } for _, test := range tests { diff --git a/core/ssz_test.go b/core/ssz_test.go index 98b10ed62..2f483fac7 100644 --- a/core/ssz_test.go +++ b/core/ssz_test.go @@ -88,6 +88,7 @@ func TestSSZ(t *testing.T) { {zero: func() any { return new(core.SyncContribution) }}, {zero: func() any { return new(core.VersionedPayloadAttestationData) }}, {zero: func() any { return new(core.VersionedPayloadAttestationMessage) }}, + {zero: func() any { return new(core.SignedProposerPreferences) }}, } f := testutil.NewEth2Fuzzer(t, 0) diff --git a/core/testdata/TestJSONSerialisation_SignedProposerPreferences.json.golden b/core/testdata/TestJSONSerialisation_SignedProposerPreferences.json.golden new file mode 100644 index 000000000..d60462e50 --- /dev/null +++ b/core/testdata/TestJSONSerialisation_SignedProposerPreferences.json.golden @@ -0,0 +1,10 @@ +{ + "message": { + "dependent_root": "0x760c58485fe3352bc5b52456f0eff11f99ef25258727c8b2922e22f5563d3949", + "proposal_slot": "16546581078090080517", + "validator_index": "18058059291253152324", + "fee_recipient": "0xd5ce92f83e0212d3077d0549b01b83b78e64fc14", + "target_gas_limit": "7556123538639849854" + }, + "signature": "0x03f2ecd45fac9d4aef1d6a649ec7799f34798eb91ff4c4fe748e21819c51945e7ed3df1fb0b5a4e0e895804813d1f4a80680595fb94b57636910c8ea580e905fda113b925adfac8d80ebb68deac0c0d6cb9aa138c60f34218fefbffdf6e04b99" +} \ No newline at end of file diff --git a/core/testdata/TestSSZSerialisation_SignedProposerPreferences.ssz.golden b/core/testdata/TestSSZSerialisation_SignedProposerPreferences.ssz.golden new file mode 100644 index 000000000..fc3bc9f6b --- /dev/null +++ b/core/testdata/TestSSZSerialisation_SignedProposerPreferences.ssz.golden @@ -0,0 +1 @@ +v XH_5+ŵ$V%%'Ȳ."V=9Ia®CD?Β>}Id~Sh_Jjdy4yt!Q^~蕀HY_KWciX_;Z߬붍˚84!K \ No newline at end of file diff --git a/core/types.go b/core/types.go index 6d0a228c0..50a88a302 100644 --- a/core/types.go +++ b/core/types.go @@ -45,9 +45,10 @@ const ( DutySyncContribution DutyType = 12 DutyInfoSync DutyType = 13 DutyPayloadAttestation DutyType = 14 + DutyProposerPreferences DutyType = 15 // Only ever append new types here... - dutySentinel DutyType = 15 // Must always be last + dutySentinel DutyType = 16 // Must always be last ) func (d DutyType) Valid() bool { @@ -71,6 +72,7 @@ func (d DutyType) String() string { DutySyncContribution: "sync_contribution", DutyInfoSync: "info_sync", DutyPayloadAttestation: "payload_attestation", + DutyProposerPreferences: "proposer_preferences", }[d] } @@ -271,6 +273,16 @@ func NewPayloadAttestationDuty(slot uint64) Duty { } } +// NewProposerPreferencesDuty returns a new proposer preferences duty, keyed by the proposal slot. +// It is a convenience function that is slightly more readable and concise than the struct literal +// equivalent. +func NewProposerPreferencesDuty(slot uint64) Duty { + return Duty{ + Slot: slot, + Type: DutyProposerPreferences, + } +} + const ( pkLen = 98 // "0x" + hex.Encode([48]byte) = 2+2*48 sigLen = 96 diff --git a/core/types_test.go b/core/types_test.go index f702b0f22..812910134 100644 --- a/core/types_test.go +++ b/core/types_test.go @@ -33,9 +33,10 @@ func TestBackwardsCompatibility(t *testing.T) { require.EqualValues(t, 12, core.DutySyncContribution) require.EqualValues(t, 13, core.DutyInfoSync) require.EqualValues(t, 14, core.DutyPayloadAttestation) + require.EqualValues(t, 15, core.DutyProposerPreferences) // Add more types here. - const sentinel = core.DutyType(15) + const sentinel = core.DutyType(16) for i := core.DutyUnknown; i <= sentinel; i++ { switch i { case core.DutyUnknown: @@ -79,7 +80,7 @@ func TestWithDutySpanCtx(t *testing.T) { func TestAllDutyTypes(t *testing.T) { adt := core.AllDutyTypes() - require.Len(t, adt, 14) + require.Len(t, adt, 15) for i, dt := range adt { require.Equal(t, i, slices.Index(adt, dt)) diff --git a/core/validatorapi/metrics.go b/core/validatorapi/metrics.go index 5a6fef5f0..e0279e18c 100644 --- a/core/validatorapi/metrics.go +++ b/core/validatorapi/metrics.go @@ -49,8 +49,19 @@ var ( Name: "vc_user_agent", Help: "Gauge with label set to user agent string of requests made by VC", }, []string{"user_agent"}) + + proposerPrefMismatch = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "core", + Subsystem: "validatorapi", + Name: "proposer_preferences_mismatch_total", + Help: "Total number of proposer preferences submitted by the VC that mismatch the cluster-lock value, by field", + }, []string{"field"}) ) +func incProposerPrefMismatch(field string) { + proposerPrefMismatch.WithLabelValues(field).Inc() +} + func incAPIErrors(endpoint string, statusCode int) { apiErrors.WithLabelValues(endpoint, strconv.Itoa(statusCode)).Inc() } diff --git a/core/validatorapi/mocks/handler.go b/core/validatorapi/mocks/handler.go index ba8e32cc5..9e2f2673f 100644 --- a/core/validatorapi/mocks/handler.go +++ b/core/validatorapi/mocks/handler.go @@ -10,6 +10,8 @@ import ( context "context" + gloas "github.com/attestantio/go-eth2-client/spec/gloas" + http "net/http" mock "github.com/stretchr/testify/mock" @@ -454,6 +456,24 @@ func (_m *Handler) SubmitProposal(ctx context.Context, opts *api.SubmitProposalO return r0 } +// SubmitProposerPreferences provides a mock function with given fields: ctx, preferences +func (_m *Handler) SubmitProposerPreferences(ctx context.Context, preferences []*gloas.SignedProposerPreferences) error { + ret := _m.Called(ctx, preferences) + + if len(ret) == 0 { + panic("no return value specified for SubmitProposerPreferences") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, []*gloas.SignedProposerPreferences) error); ok { + r0 = rf(ctx, preferences) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // SubmitSyncCommitteeContributions provides a mock function with given fields: ctx, contributionAndProofs func (_m *Handler) SubmitSyncCommitteeContributions(ctx context.Context, contributionAndProofs []*altair.SignedContributionAndProof) error { ret := _m.Called(ctx, contributionAndProofs) diff --git a/core/validatorapi/router.go b/core/validatorapi/router.go index 4b7623aa6..17b1a9a50 100644 --- a/core/validatorapi/router.go +++ b/core/validatorapi/router.go @@ -92,6 +92,10 @@ type Handler interface { eth2client.VoluntaryExitSubmitter // Above sorted alphabetically. + // SubmitProposerPreferences receives partially signed proposer preferences from the validator client. + // TODO(gloas): replace with eth2client.ProposerPreferencesSubmitter once attestantio/go-eth2-client#316 merges. + SubmitProposerPreferences(ctx context.Context, preferences []*gloas.SignedProposerPreferences) error + // Address returns the address of the beacon node. Address() string // Headers returns custom headers to include in requests to the beacon node. @@ -109,6 +113,9 @@ func NewRouter(h Handler, builderEnabled bool) (*mux.Router, error) { Handler handlerFunc Methods []string Encodings []contentType + // MaxBody limits the request body size in bytes, enforced at the read boundary + // before the body is buffered; zero means unlimited. + MaxBody int64 }{ { Name: "attester_duties", @@ -334,6 +341,14 @@ func NewRouter(h Handler, builderEnabled bool) (*mux.Router, error) { Methods: []string{http.MethodPost}, Encodings: []contentType{contentTypeJSON}, }, + { + Name: "submit_proposer_preferences", + Path: "/eth/v1/validator/proposer_preferences", + Handler: submitProposerPreferences(h), + Methods: []string{http.MethodPost}, + Encodings: []contentType{contentTypeJSON, contentTypeSSZ}, + MaxBody: maxProposerPreferencesBody, + }, { Name: "aggregate_sync_committee_selections", Path: "/eth/v1/validator/sync_committee_selections", @@ -352,7 +367,7 @@ func NewRouter(h Handler, builderEnabled bool) (*mux.Router, error) { r := mux.NewRouter() for _, e := range endpoints { - handler := r.Handle(e.Path, wrap(e.Name, e.Handler, e.Encodings)) + handler := r.Handle(e.Path, wrap(e.Name, e.Handler, e.Encodings, e.MaxBody)) if len(e.Methods) != 0 { handler.Methods(e.Methods...) } @@ -381,13 +396,26 @@ func (a apiError) Error() string { return fmt.Sprintf("api error[status=%d,msg=%s]: %v", a.StatusCode, a.Message, a.Err) } +// badRequestError returns an apiError with status 400 for client input validation failures. +func badRequestError(msg string, err error) error { + if err == nil { + err = errors.New(msg) + } + + return apiError{ + StatusCode: http.StatusBadRequest, + Message: msg, + Err: err, + } +} + // handlerFunc is a convenient handler function providing a context, parsed path parameters, // the request body, and returning the response struct or an error. type handlerFunc func(ctx context.Context, params map[string]string, header http.Header, query url.Values, typ contentType, body []byte) (res any, headers http.Header, err error) // wrap adapts the handler function returning a standard http handler. // It does tracing, metrics and response and error writing. -func wrap(endpoint string, handler handlerFunc, encodings []contentType) http.Handler { +func wrap(endpoint string, handler handlerFunc, encodings []contentType, maxBody int64) http.Handler { wrap := func(w http.ResponseWriter, r *http.Request) { ctx := r.Context() ctx = log.WithTopic(ctx, "vapi") @@ -443,9 +471,23 @@ func wrap(endpoint string, handler handlerFunc, encodings []contentType) http.Ha recordVCUserAgent(userAgent) } + if maxBody > 0 { + // Enforce the endpoint body limit at the read boundary, before buffering. + r.Body = http.MaxBytesReader(w, r.Body, maxBody) + } + body, err := io.ReadAll(r.Body) if err != nil { + if maxBytesErr := new(http.MaxBytesError); errors.As(err, &maxBytesErr) { + err = apiError{ + StatusCode: http.StatusRequestEntityTooLarge, + Message: "request body too large", + Err: err, + } + } + writeError(ctx, w, endpoint, err) + return } @@ -1802,6 +1844,88 @@ func submitSyncCommitteeMessages(s eth2client.SyncCommitteeMessagesSubmitter) ha } } +// maxProposerPreferencesBody is a sanity cap on the proposer preferences request body size, +// enforced by the route's MaxBody at the read boundary. It is far above the encoded size of +// the spec list limit ((MIN_SEED_LOOKAHEAD+1)*SLOTS_PER_EPOCH items) on any network; the +// exact spec limit is enforced in Component.SubmitProposerPreferences. +const maxProposerPreferencesBody = 1 << 20 // 1MB + +// submitProposerPreferences receives partially signed proposer preferences from the validator +// client and forwards them for threshold aggregation. From the gloas fork, these supersede +// prepare_beacon_proposer and register_validator as the source of fee recipient and gas limit. +func submitProposerPreferences(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 prefs []*gloas.SignedProposerPreferences + + if typ == contentTypeSSZ { + prefs, err = unmarshalProposerPreferencesSSZ(body) + } else { + err = unmarshal(typ, body, &prefs) + } + + if err != nil { + return nil, nil, errors.Wrap(err, "unmarshal proposer preferences") + } + + err = h.SubmitProposerPreferences(ctx, prefs) + if err != nil { + return nil, nil, err + } + + return nil, nil, nil + } +} + +// unmarshalProposerPreferencesSSZ decodes an SSZ List[SignedProposerPreferences]. The element +// type is fixed-size, so the list is encoded as plain concatenation without an offset table. +// An empty body is a valid empty list, matching the JSON `[]` behavior. +func unmarshalProposerPreferencesSSZ(body []byte) ([]*gloas.SignedProposerPreferences, error) { + itemSize := (&gloas.SignedProposerPreferences{Message: &gloas.ProposerPreferences{}}).SizeSSZ() + + if len(body)%itemSize != 0 { + return nil, apiError{ + StatusCode: http.StatusBadRequest, + Message: "invalid ssz proposer preferences list length", + Err: errors.New("invalid ssz list length", z.Int("length", len(body)), z.Int("item_size", itemSize)), + } + } + + prefs := make([]*gloas.SignedProposerPreferences, 0, len(body)/itemSize) + + for i := 0; i < len(body); i += itemSize { + pref := new(gloas.SignedProposerPreferences) + if err := pref.UnmarshalSSZ(body[i : i+itemSize]); err != nil { + return nil, apiError{ + StatusCode: http.StatusBadRequest, + Message: "failed parsing ssz proposer preferences", + Err: err, + } + } + + prefs = append(prefs, pref) + } + + return prefs, nil +} + // submitProposalPreparations swallows fee-recipient-address from validator client as it should be // configured by charon from cluster-lock.json and VC need not be configured with correct fee-recipient-address. func submitProposalPreparations() handlerFunc { diff --git a/core/validatorapi/router_internal_test.go b/core/validatorapi/router_internal_test.go index 4eab33085..4e7b18b3c 100644 --- a/core/validatorapi/router_internal_test.go +++ b/core/validatorapi/router_internal_test.go @@ -2368,6 +2368,7 @@ type testHandler struct { PayloadAttestationDataFunc func(ctx context.Context, opts *eth2api.PayloadAttestationDataOpts) (*eth2api.Response[*eth2spec.VersionedPayloadAttestationData], error) PTCDutiesFunc func(ctx context.Context, opts *eth2api.PTCDutiesOpts) (*eth2api.Response[[]*eth2v1.PTCDuty], error) SubmitPayloadAttMsgsFunc func(ctx context.Context, opts *eth2api.SubmitPayloadAttestationMessagesOpts) error + SubmitProposerPreferencesFunc func(ctx context.Context, preferences []*gloas.SignedProposerPreferences) error ProxyFunc func(ctx context.Context, req *http.Request) (*http.Response, error) AddressFunc func() string HeadersFunc func() map[string]string @@ -2385,6 +2386,10 @@ func (h testHandler) SubmitPayloadAttestationMessages(ctx context.Context, opts return h.SubmitPayloadAttMsgsFunc(ctx, opts) } +func (h testHandler) SubmitProposerPreferences(ctx context.Context, preferences []*gloas.SignedProposerPreferences) error { + return h.SubmitProposerPreferencesFunc(ctx, preferences) +} + func (h testHandler) PTCDuties(ctx context.Context, opts *eth2api.PTCDutiesOpts) (*eth2api.Response[[]*eth2v1.PTCDuty], error) { return h.PTCDutiesFunc(ctx, opts) } @@ -2793,3 +2798,132 @@ func TestPayloadAttestationRoutes(t *testing.T) { testRawRouter(t, handler, callback) }) } + +func TestUnmarshalProposerPreferencesSSZ(t *testing.T) { + pref1 := testutil.RandomProposerPreferences() + pref2 := testutil.RandomProposerPreferences() + + b1, err := pref1.MarshalSSZ() + require.NoError(t, err) + b2, err := pref2.MarshalSSZ() + require.NoError(t, err) + + prefs, err := unmarshalProposerPreferencesSSZ(append(b1, b2...)) + require.NoError(t, err) + require.Len(t, prefs, 2) + require.Equal(t, pref1, prefs[0]) + require.Equal(t, pref2, prefs[1]) + + // An empty body is a valid empty list. + prefs, err = unmarshalProposerPreferencesSSZ(nil) + require.NoError(t, err) + require.Empty(t, prefs) + + // Invalid lengths are rejected. + _, err = unmarshalProposerPreferencesSSZ(b1[:len(b1)-1]) + require.Error(t, err) +} + +func TestSubmitProposerPreferencesRouter(t *testing.T) { + prefs := []*gloas.SignedProposerPreferences{testutil.RandomProposerPreferences(), testutil.RandomProposerPreferences()} + + newHandler := func(submitted *[][]*gloas.SignedProposerPreferences) testHandler { + return testHandler{ + SubmitProposerPreferencesFunc: func(_ context.Context, preferences []*gloas.SignedProposerPreferences) error { + *submitted = append(*submitted, preferences) + + return nil + }, + } + } + + post := func(ctx context.Context, t *testing.T, baseURL, contentType, version string, body []byte) *http.Response { + t.Helper() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/eth/v1/validator/proposer_preferences", bytes.NewReader(body)) + require.NoError(t, err) + + req.Header.Set("Content-Type", contentType) + if version != "" { + req.Header.Set(versionHeader, version) + } + + res, err := new(http.Client).Do(req) + require.NoError(t, err) + + return res + } + + t.Run("json", func(t *testing.T) { + var submitted [][]*gloas.SignedProposerPreferences + + body, err := json.Marshal(prefs) + require.NoError(t, err) + + testRawRouter(t, newHandler(&submitted), func(ctx context.Context, baseURL string) { + res := post(ctx, t, baseURL, "application/json", "gloas", body) + require.Equal(t, http.StatusOK, res.StatusCode) + require.Len(t, submitted, 1) + require.Equal(t, prefs, submitted[0]) + }) + }) + + t.Run("ssz", func(t *testing.T) { + var submitted [][]*gloas.SignedProposerPreferences + + var body []byte + for _, pref := range prefs { + b, err := pref.MarshalSSZ() + require.NoError(t, err) + + body = append(body, b...) + } + + testRawRouter(t, newHandler(&submitted), func(ctx context.Context, baseURL string) { + res := post(ctx, t, baseURL, "application/octet-stream", "gloas", body) + require.Equal(t, http.StatusOK, res.StatusCode) + require.Len(t, submitted, 1) + require.Equal(t, prefs, submitted[0]) + }) + }) + + t.Run("missing version header", func(t *testing.T) { + var submitted [][]*gloas.SignedProposerPreferences + + testRawRouter(t, newHandler(&submitted), func(ctx context.Context, baseURL string) { + res := post(ctx, t, baseURL, "application/json", "", []byte("[]")) + require.Equal(t, http.StatusBadRequest, res.StatusCode) + require.Empty(t, submitted) + }) + }) + + t.Run("wrong version header", func(t *testing.T) { + var submitted [][]*gloas.SignedProposerPreferences + + testRawRouter(t, newHandler(&submitted), func(ctx context.Context, baseURL string) { + res := post(ctx, t, baseURL, "application/json", "electra", []byte("[]")) + require.Equal(t, http.StatusBadRequest, res.StatusCode) + require.Empty(t, submitted) + }) + }) + + t.Run("invalid ssz length", func(t *testing.T) { + var submitted [][]*gloas.SignedProposerPreferences + + testRawRouter(t, newHandler(&submitted), func(ctx context.Context, baseURL string) { + res := post(ctx, t, baseURL, "application/octet-stream", "gloas", []byte{0x01, 0x02}) + require.Equal(t, http.StatusBadRequest, res.StatusCode) + require.Empty(t, submitted) + }) + }) + + t.Run("oversized body", func(t *testing.T) { + var submitted [][]*gloas.SignedProposerPreferences + + testRawRouter(t, newHandler(&submitted), func(ctx context.Context, baseURL string) { + res := post(ctx, t, baseURL, "application/octet-stream", "gloas", make([]byte, maxProposerPreferencesBody+1)) + require.Equal(t, http.StatusRequestEntityTooLarge, res.StatusCode) + require.Empty(t, submitted) + }) + }) +} diff --git a/core/validatorapi/validatorapi.go b/core/validatorapi/validatorapi.go index 7a2123134..3a6648e73 100644 --- a/core/validatorapi/validatorapi.go +++ b/core/validatorapi/validatorapi.go @@ -9,6 +9,7 @@ import ( "math/big" "net/http" "runtime" + "strings" "testing" "time" @@ -16,6 +17,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" @@ -982,6 +984,142 @@ func (c Component) SubmitSyncCommitteeMessages(ctx context.Context, messages []* return nil } +// SubmitProposerPreferences receives partially signed gloas.SignedProposerPreferences from the +// validator client, verifies each partial signature and forwards them to subscribers grouped by +// proposal slot for threshold aggregation. Preferences are aggregated ungated (like sync committee +// messages): the call never waits for other shares; the submission completing the threshold +// synchronously triggers aggregation, like other VC-pushed duties. +// +// TODO(gloas): swap for eth2client.ProposerPreferencesSubmitter once attestantio/go-eth2-client#316 +// merges. +func (c Component) SubmitProposerPreferences(ctx context.Context, preferences []*gloas.SignedProposerPreferences) error { + // Use complete validators since preferences are submitted ahead of time: a validator + // activating in the proposal epoch is a valid proposer but not yet active when submitting. + vals, err := c.eth2Cl.CompleteValidators(ctx) + if err != nil { + return err + } + + currentSlot, err := SlotFromTimestamp(ctx, c.eth2Cl, time.Now()) + if err != nil { + return err + } + + _, slotsPerEpoch, err := eth2wrap.FetchSlotsConfig(ctx, c.eth2Cl) + if err != nil { + return err + } + + // MIN_SEED_LOOKAHEAD as per the consensus spec phase0 preset (identical on all networks). + const minSeedLookahead = 1 + + // The request body is defined as List[SignedProposerPreferences, (MIN_SEED_LOOKAHEAD + 1) * SLOTS_PER_EPOCH]. + // Enforce the list limit before any signature verification to bound memory and CPU. + if uint64(len(preferences)) > (minSeedLookahead+1)*slotsPerEpoch { + return badRequestError("too many proposer preferences", + errors.New("too many proposer preferences", z.Int("count", len(preferences)), z.U64("limit", (minSeedLookahead+1)*slotsPerEpoch))) + } + + // Preferences are only valid for future proposal slots within the proposer lookahead: + // the current epoch up to MIN_SEED_LOOKAHEAD epochs ahead. Rejecting the rest bounds + // the entries stored and exchanged with peers per valid share. + maxSlot := eth2p0.Slot((uint64(currentSlot)/slotsPerEpoch + 1 + minSeedLookahead) * slotsPerEpoch) + + psigsBySlot := make(map[eth2p0.Slot]core.ParSignedDataSet) + + // Faulty entries are skipped with a warning instead of failing the whole batch: entries + // target independent proposal slots, and the list limit above already bounds the work. + for _, pref := range preferences { + if pref == nil || pref.Message == nil { + log.Warn(ctx, "Skipping nil proposer preferences message", nil) + continue + } + + slot := pref.Message.ProposalSlot + if slot <= currentSlot || slot >= maxSlot { + log.Warn(ctx, "Skipping proposer preferences with proposal slot outside lookahead window", nil, + z.U64("proposal_slot", uint64(slot)), z.U64("current_slot", uint64(currentSlot)), z.U64("max_slot", uint64(maxSlot))) + continue + } + + val, ok := vals[pref.Message.ValidatorIndex] + if !ok || val.Validator == nil { + log.Warn(ctx, "Skipping proposer preferences for unknown validator", nil, + z.U64("validator_index", uint64(pref.Message.ValidatorIndex))) + continue + } + + pk, err := core.PubKeyFromBytes(val.Validator.PublicKey[:]) + if err != nil { + return err + } + + parSigData := core.NewPartialSignedProposerPreferences(pref, c.shareIdx) + + err = c.verifyPartialSig(ctx, parSigData, pk) + if err != nil { + log.Warn(ctx, "Skipping proposer preferences with invalid partial signature", err, + z.U64("proposal_slot", uint64(slot)), z.Any("pubkey", pk)) + continue + } + + c.warnProposerPreferencesMismatch(ctx, pk, pref.Message) + + log.Debug(ctx, "Proposer preferences received from validator client", + z.U64("proposal_slot", uint64(slot)), + z.Str("fee_recipient", fmt.Sprintf("%#x", pref.Message.FeeRecipient)), + z.U64("target_gas_limit", pref.Message.TargetGasLimit)) + + if _, ok := psigsBySlot[slot]; !ok { + psigsBySlot[slot] = make(core.ParSignedDataSet) + } + + psigsBySlot[slot][pk] = parSigData + } + + for slot, data := range psigsBySlot { + duty := core.NewProposerPreferencesDuty(uint64(slot)) + for _, sub := range c.subs { + err = sub(ctx, duty, data) + if err != nil { + return err + } + } + } + + return nil +} + +// warnProposerPreferencesMismatch logs a warning and increments a metric when the VC-submitted fee +// recipient or target gas limit differs from the cluster-lock value. It does not reject the +// preference: aggregation proceeds on whatever value reaches threshold (so a staggered gas-limit +// change self-heals); the warning surfaces operator misconfiguration. +func (c Component) warnProposerPreferencesMismatch(ctx context.Context, pubkey core.PubKey, msg *gloas.ProposerPreferences) { + if c.feeRecipientFunc == nil { + return + } + + expectedFeeRecipient := c.feeRecipientFunc(pubkey) + actualFeeRecipient := fmt.Sprintf("%#x", msg.FeeRecipient) + + if !strings.EqualFold(actualFeeRecipient, expectedFeeRecipient) { + log.Warn(ctx, "Proposer preferences with unexpected fee recipient", nil, + z.Any("pubkey", pubkey), + z.Str("expected", expectedFeeRecipient), + z.Str("actual", actualFeeRecipient)) + incProposerPrefMismatch("fee_recipient") + } + + // TargetGasLimit is only enforced for cluster-lock versions that support it (non-zero). + if c.targetGasLimit != 0 && msg.TargetGasLimit != uint64(c.targetGasLimit) { + log.Warn(ctx, "Proposer preferences with unexpected target gas limit", nil, + z.Any("pubkey", pubkey), + z.U64("expected", uint64(c.targetGasLimit)), + z.U64("actual", msg.TargetGasLimit)) + incProposerPrefMismatch("gas_limit") + } +} + // PayloadAttestationData implements the eth2client.PayloadAttestationDataProvider for the router. func (c Component) PayloadAttestationData(ctx context.Context, opts *eth2api.PayloadAttestationDataOpts) (*eth2api.Response[*eth2spec.VersionedPayloadAttestationData], error) { var span trace.Span diff --git a/core/validatorapi/validatorapi_internal_test.go b/core/validatorapi/validatorapi_internal_test.go index 3e99b63f2..3102c84d1 100644 --- a/core/validatorapi/validatorapi_internal_test.go +++ b/core/validatorapi/validatorapi_internal_test.go @@ -3,9 +3,14 @@ package validatorapi import ( + "context" + "fmt" "testing" + "github.com/attestantio/go-eth2-client/spec/bellatrix" + "github.com/attestantio/go-eth2-client/spec/gloas" eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0" + promtestutil "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" "github.com/obolnetwork/charon/core" @@ -95,3 +100,38 @@ func TestWrapResponseWithMetadata(t *testing.T) { require.Equal(t, 123, resp.Data) require.Equal(t, metadata, resp.Metadata) } + +func TestWarnProposerPreferencesMismatch(t *testing.T) { + ctx := context.Background() + pubkey := testutil.RandomCorePubKey(t) + + addr := bellatrix.ExecutionAddress{0x01, 0x02, 0x03} + feeRecipient := fmt.Sprintf("%#x", addr) + + const gasLimit = 30000000 + + c := Component{ + feeRecipientFunc: func(core.PubKey) string { return feeRecipient }, + targetGasLimit: gasLimit, + } + + feeCount := func() float64 { return promtestutil.ToFloat64(proposerPrefMismatch.WithLabelValues("fee_recipient")) } + gasCount := func() float64 { return promtestutil.ToFloat64(proposerPrefMismatch.WithLabelValues("gas_limit")) } + + fee0, gas0 := feeCount(), gasCount() + + // Matching fee recipient and gas limit: no increment. + c.warnProposerPreferencesMismatch(ctx, pubkey, &gloas.ProposerPreferences{FeeRecipient: addr, TargetGasLimit: gasLimit}) + require.InDelta(t, fee0, feeCount(), 0) + require.InDelta(t, gas0, gasCount(), 0) + + // Mismatching fee recipient and gas limit: both increment. + otherAddr := bellatrix.ExecutionAddress{0xff} + c.warnProposerPreferencesMismatch(ctx, pubkey, &gloas.ProposerPreferences{FeeRecipient: otherAddr, TargetGasLimit: gasLimit + 1}) + require.InDelta(t, fee0+1, feeCount(), 0) + require.InDelta(t, gas0+1, gasCount(), 0) + + // Nil feeRecipientFunc: no panic, no increment. + empty := Component{} + empty.warnProposerPreferencesMismatch(ctx, pubkey, &gloas.ProposerPreferences{FeeRecipient: otherAddr}) +} diff --git a/core/validatorapi/validatorapi_test.go b/core/validatorapi/validatorapi_test.go index 826a23422..6efc82f12 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" @@ -2057,6 +2058,76 @@ func TestComponent_SubmitSyncCommitteeMessages(t *testing.T) { require.Equal(t, count, 1) } +func TestComponent_SubmitProposerPreferences(t *testing.T) { + const vIdx = 1 + + var ( + ctx = context.Background() + pref = testutil.RandomProposerPreferences() + pubkey = beaconmock.ValidatorSetA[vIdx].Validator.PublicKey + count = 0 // No of times the subscription function is called. + ) + + pref.Message.ValidatorIndex = vIdx + + bmock, err := beaconmock.New(t.Context(), beaconmock.WithValidatorSet(beaconmock.ValidatorSetA)) + require.NoError(t, err) + + // Preferences are only accepted for future slots within the proposer lookahead. + currentSlot, err := validatorapi.SlotFromTimestamp(ctx, bmock, time.Now()) + require.NoError(t, err) + + pref.Message.ProposalSlot = currentSlot + 1 + + 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.NewProposerPreferencesDuty(uint64(pref.Message.ProposalSlot)), duty) + + pk, err := core.PubKeyFromBytes(pubkey[:]) + require.NoError(t, err) + + data, ok := set[pk] + require.True(t, ok) + require.Equal(t, core.NewPartialSignedProposerPreferences(pref, 0), data) + + count++ + + return nil + }) + + require.NoError(t, vapi.SubmitProposerPreferences(ctx, []*gloas.SignedProposerPreferences{pref})) + require.Equal(t, count, 1) + + // Faulty entries are skipped with a warning, not forwarded and not failing the batch: + // already-started slots, far-future slots, unknown validators and nil entries. + pref.Message.ProposalSlot = currentSlot + require.NoError(t, vapi.SubmitProposerPreferences(ctx, []*gloas.SignedProposerPreferences{pref})) + + pref.Message.ProposalSlot = currentSlot + 100_000 + require.NoError(t, vapi.SubmitProposerPreferences(ctx, []*gloas.SignedProposerPreferences{pref})) + + pref.Message.ProposalSlot = currentSlot + 1 + pref.Message.ValidatorIndex = 99_999 + require.NoError(t, vapi.SubmitProposerPreferences(ctx, []*gloas.SignedProposerPreferences{pref})) + + require.NoError(t, vapi.SubmitProposerPreferences(ctx, []*gloas.SignedProposerPreferences{nil})) + + require.Equal(t, count, 1) // Subscriber not called for any skipped entry. + + // Requests exceeding the spec list limit are rejected. + _, slotsPerEpoch, err := eth2wrap.FetchSlotsConfig(ctx, bmock) + require.NoError(t, err) + + tooMany := make([]*gloas.SignedProposerPreferences, 2*slotsPerEpoch+1) + for i := range tooMany { + tooMany[i] = testutil.RandomProposerPreferences() + } + + require.ErrorContains(t, vapi.SubmitProposerPreferences(ctx, tooMany), "too many proposer preferences") +} + func TestComponent_SubmitSyncCommitteeContributions(t *testing.T) { const vIdx = 1 diff --git a/docs/metrics.md b/docs/metrics.md index fcc545398..f1398405d 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -97,6 +97,7 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance. | `core_tracker_participation_total` | Counter | Total number of successful participations by peer and duty type | `duty, peer` | | `core_tracker_success_duties_total` | Counter | Total number of successful duties by type | `duty` | | `core_tracker_unexpected_events_total` | Counter | Total number of unexpected events by peer | `peer` | +| `core_validatorapi_proposer_preferences_mismatch_total` | Counter | Total number of proposer preferences submitted by the VC that mismatch the cluster-lock value, by field | `field` | | `core_validatorapi_proxy_request_latency_seconds` | Histogram | The validatorapi proxy request latencies in seconds by path | `path` | | `core_validatorapi_request_error_total` | Counter | The total number of validatorapi request errors | `endpoint, status_code` | | `core_validatorapi_request_latency_seconds` | Histogram | The validatorapi request latencies in seconds by endpoint | `endpoint` | diff --git a/eth2util/signing/signing.go b/eth2util/signing/signing.go index c96d7a0f5..6703d415f 100644 --- a/eth2util/signing/signing.go +++ b/eth2util/signing/signing.go @@ -33,6 +33,7 @@ const ( DomainDeposit DomainName = "DOMAIN_DEPOSIT" DomainBlobSidecar DomainName = "DOMAIN_BLOB_SIDECAR" DomainPTCAttester DomainName = "DOMAIN_PTC_ATTESTER" + DomainProposerPreferences DomainName = "DOMAIN_PROPOSER_PREFERENCES" ) // GetDomain returns the beacon domain for the provided type. diff --git a/go.mod b/go.mod index 2bd02d745..361d79d75 100644 --- a/go.mod +++ b/go.mod @@ -289,7 +289,7 @@ require ( replace github.com/coinbase/kryptology => github.com/ObolNetwork/kryptology v0.1.0 // We're replacing go-eth2-client with a branch off our fork. The branch is kept up to date with the latest attestantio versions. -replace github.com/attestantio/go-eth2-client => github.com/ObolNetwork/go-eth2-client v0.29.0-obol.2-gloas //nolint +replace github.com/attestantio/go-eth2-client => github.com/ObolNetwork/go-eth2-client v0.29.0-obol.3-gloas //nolint tool ( github.com/bufbuild/buf/cmd/buf diff --git a/go.sum b/go.sum index 520218fe6..49cf41247 100644 --- a/go.sum +++ b/go.sum @@ -42,8 +42,8 @@ github.com/DataDog/zstd v1.5.7 h1:ybO8RBeh29qrxIhCA9E8gKY6xfONU9T6G6aP9DTKfLE= github.com/DataDog/zstd v1.5.7/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= -github.com/ObolNetwork/go-eth2-client v0.29.0-obol.2-gloas h1:zWRrvhl/GHiTXTgeVk7ikTO4VMHEQy8kgHha8yr8R74= -github.com/ObolNetwork/go-eth2-client v0.29.0-obol.2-gloas/go.mod h1:yhVnKAzIsFhtawbq6k/rA/Dy4vsPpu2Z2cGdQVrIjd0= +github.com/ObolNetwork/go-eth2-client v0.29.0-obol.3-gloas h1:6BNDbYWODMTbb3hc/GeGgdNzHd3ZJUBVpJKrQhvY7bU= +github.com/ObolNetwork/go-eth2-client v0.29.0-obol.3-gloas/go.mod h1:yhVnKAzIsFhtawbq6k/rA/Dy4vsPpu2Z2cGdQVrIjd0= github.com/ObolNetwork/kryptology v0.1.0 h1:AhoG4My70+xMhEJSpVaJay/t+T/vIUNHQYLjsDJHulI= github.com/ObolNetwork/kryptology v0.1.0/go.mod h1:/Wl7Js2f676GyXZDTaojf/O+l0fxFPWudbyjdFhkpSA= github.com/OffchainLabs/go-bitfield v0.0.0-20251031151322-f427d04d8506 h1:d/SJkN8/9Ca+1YmuDiUJxAiV4w/a9S8NcsG7GMQSrVI= diff --git a/testutil/beaconmock/static.json b/testutil/beaconmock/static.json index 91a18fd39..175b92a61 100644 --- a/testutil/beaconmock/static.json +++ b/testutil/beaconmock/static.json @@ -100,6 +100,7 @@ "TARGET_AGGREGATORS_PER_COMMITTEE": "16", "DOMAIN_SYNC_COMMITTEE_SELECTION_PROOF": "0x08000000", "DOMAIN_PTC_ATTESTER": "0x0c000000", + "DOMAIN_PROPOSER_PREFERENCES": "0x0d000000", "MESSAGE_DOMAIN_INVALID_SNAPPY": "0x00000000", "EPOCHS_PER_SLASHINGS_VECTOR": "8192", "MIN_SLASHING_PENALTY_QUOTIENT": "128", diff --git a/testutil/random.go b/testutil/random.go index 45df5fd43..789dc9fe9 100644 --- a/testutil/random.go +++ b/testutil/random.go @@ -1061,6 +1061,19 @@ func RandomVersionedPayloadAttestationMessage() *eth2spec.VersionedPayloadAttest } } +func RandomProposerPreferences() *gloas.SignedProposerPreferences { + return &gloas.SignedProposerPreferences{ + Message: &gloas.ProposerPreferences{ + DependentRoot: RandomRoot(), + ProposalSlot: RandomSlot(), + ValidatorIndex: RandomVIdx(), + FeeRecipient: RandomExecutionAddress(), + TargetGasLimit: rand.Uint64(), + }, + Signature: RandomEth2Signature(), + } +} + func RandomSyncCommitteeMessage() *altair.SyncCommitteeMessage { return &altair.SyncCommitteeMessage{ Slot: RandomSlot(),