Skip to content

adding the duties stores for validator duties processing #16479

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
james-prysm merged 22 commits into from
Mar 9, 2026
Merged

adding the duties stores for validator duties processing #16479

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
a38a2e0
adding the duties stores, useful for using the split endpoint in anot…
james-prysm Mar 5, 2026
7593314
gaz
james-prysm Mar 5, 2026
0723fba
Merge branch 'develop' into duties-store
james-prysm Mar 5, 2026
d377ca5
changelog
james-prysm Mar 5, 2026
6a6028b
changelog
james-prysm Mar 5, 2026
7c05229
fixing changelog comment
james-prysm Mar 5, 2026
366998b
fixing test
james-prysm Mar 5, 2026
09b2d8d
Merge branch 'develop' into duties-store
james-prysm Mar 5, 2026
6c9878e
bastin's comment
james-prysm Mar 6, 2026
053fd25
fixing variable
james-prysm Mar 6, 2026
57a4b86
Merge branch 'develop' into duties-store
james-prysm Mar 6, 2026
580cd8d
Merge branch 'develop' into duties-store
james-prysm Mar 6, 2026
41b4c3a
Update validator/client/duties.go
james-prysm Mar 6, 2026
6c378a2
preston's comment
james-prysm Mar 6, 2026
e6fa428
Merge branch 'develop' into duties-store
james-prysm Mar 6, 2026
93fd982
Merge branch 'develop' into duties-store
james-prysm Mar 9, 2026
402ae9f
fixing logs
james-prysm Mar 9, 2026
6a7059b
adding PTC duties to store
james-prysm Mar 9, 2026
0c3e21b
Merge branch 'develop' into duties-store
james-prysm Mar 9, 2026
cf0488c
fixing tests
james-prysm Mar 9, 2026
b479ad2
gofmt
james-prysm Mar 9, 2026
407795c
Merge branch 'develop' into duties-store
james-prysm Mar 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions changelog/james-prysm_duties-store.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
### Ignored

- refactored validator duties processing to store duties in the duties store which will make split duties api easy to implement.
2 changes: 2 additions & 0 deletions validator/client/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ go_library(
"attest.go",
"distributed.go",
"duties.go",
"duty_store.go",
"health_monitor.go",
"key_reload.go",
"log.go",
Expand Down Expand Up @@ -107,6 +108,7 @@ go_test(
"aggregate_test.go",
"attest_test.go",
"duties_test.go",
"duty_store_test.go",
"health_monitor_test.go",
"key_reload_test.go",
"log_test.go",
Expand Down
46 changes: 15 additions & 31 deletions validator/client/aggregate_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ func TestSubmitAggregateAndProof_GetDutiesRequestFailure(t *testing.T) {
t.Run(fmt.Sprintf("SlashingProtectionMinimal:%v", isSlashingProtectionMinimal), func(t *testing.T) {
hook := logTest.NewGlobal()
validator, _, validatorKey, finish := setup(t, isSlashingProtectionMinimal)
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{}}
validator.duties = testDutyStore()
defer finish()

var pubKey [fieldparams.BLSPubkeyLength]byte
Expand All @@ -45,13 +45,9 @@ func TestSubmitAggregateAndProof_SignFails(t *testing.T) {
defer finish()
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{
CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
},
},
}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
})

m.validatorClient.EXPECT().DomainData(
gomock.Any(), // ctx
Expand Down Expand Up @@ -90,13 +86,9 @@ func TestSubmitAggregateAndProof_Ok(t *testing.T) {
defer finish()
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{
CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
},
},
}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
})

m.validatorClient.EXPECT().DomainData(
gomock.Any(), // ctx
Expand Down Expand Up @@ -143,13 +135,9 @@ func TestSubmitAggregateAndProof_Ok(t *testing.T) {
defer finish()
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{
CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
},
},
}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
})

m.validatorClient.EXPECT().DomainData(
gomock.Any(), // ctx
Expand Down Expand Up @@ -197,15 +185,11 @@ func TestSubmitAggregateAndProof_Distributed(t *testing.T) {

var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{
CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
ValidatorIndex: validatorIdx,
AttesterSlot: slot,
},
},
}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
ValidatorIndex: validatorIdx,
AttesterSlot: slot,
})

validator.distributed = true
validator.attSelections = make(map[attSelectionKey]iface.BeaconCommitteeSelection)
Expand Down
14 changes: 5 additions & 9 deletions validator/client/attest.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package client

import (
"bytes"
"context"
"fmt"
"strings"
Expand Down Expand Up @@ -194,17 +193,14 @@ func (v *validator) SubmitAttestation(ctx context.Context, slot primitives.Slot,
func (v *validator) duty(pubKey [fieldparams.BLSPubkeyLength]byte) (*ethpb.ValidatorDuty, error) {
v.dutiesLock.RLock()
defer v.dutiesLock.RUnlock()
if v.duties == nil {
if !v.duties.IsInitialized() {
return nil, errors.New("no duties for validators")
}

for _, duty := range v.duties.CurrentEpochDuties {
if bytes.Equal(pubKey[:], duty.PublicKey) {
return duty, nil
}
d, ok := v.duties.CurrentDuty(pubKey)
if !ok {
return nil, fmt.Errorf("pubkey %#x not in duties", bytesutil.Trunc(pubKey[:]))
}

return nil, fmt.Errorf("pubkey %#x not in duties", bytesutil.Trunc(pubKey[:]))
return d, nil
}

// Given validator's public key, this function returns the signature of an attestation data and its signing root.
Expand Down
138 changes: 61 additions & 77 deletions validator/client/attest_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ func TestRequestAttestation_ValidatorDutiesRequestFailure(t *testing.T) {
t.Run(fmt.Sprintf("SlashingProtectionMinimal:%v", isSlashingProtectionMinimal), func(t *testing.T) {
hook := logTest.NewGlobal()
validator, _, validatorKey, finish := setup(t, isSlashingProtectionMinimal)
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{}}
validator.duties = testDutyStore()
defer finish()

var pubKey [fieldparams.BLSPubkeyLength]byte
Expand All @@ -53,12 +53,11 @@ func TestAttestToBlockHead_SubmitAttestation_EmptyCommittee(t *testing.T) {
defer finish()
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 0,
ValidatorIndex: 0,
}}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 0,
ValidatorIndex: 0,
})
validator.SubmitAttestation(t.Context(), 0, pubKey)
require.LogsContain(t, hook, "Empty committee")
})
Expand All @@ -72,13 +71,12 @@ func TestAttestToBlockHead_SubmitAttestation_RequestFailure(t *testing.T) {

validator, m, validatorKey, finish := setup(t, isSlashingProtectionMinimal)
defer finish()
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: 111,
ValidatorIndex: 0,
}}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: 111,
ValidatorIndex: 0,
})
m.validatorClient.EXPECT().AttestationData(
gomock.Any(), // ctx
gomock.AssignableToTypeOf(&ethpb.AttestationDataRequest{}),
Expand Down Expand Up @@ -145,14 +143,12 @@ func TestSubmitAttestation_ElectraCommitteeIndex(t *testing.T) {
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: tt.assignedCommitteeIndex,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
},
}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: tt.assignedCommitteeIndex,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
})

var capturedRequest *ethpb.AttestationDataRequest
// Capture the actual request to verify committee index
Expand Down Expand Up @@ -207,15 +203,13 @@ func TestAttestToBlockHead_AttestsCorrectly(t *testing.T) {
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorCommitteeIndex: 4,
ValidatorIndex: validatorIndex,
},
}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorCommitteeIndex: 4,
ValidatorIndex: validatorIndex,
})

beaconBlockRoot := bytesutil.ToBytes32([]byte("A"))
targetRoot := bytesutil.ToBytes32([]byte("B"))
Expand Down Expand Up @@ -288,14 +282,12 @@ func TestAttestToBlockHead_AttestsCorrectly(t *testing.T) {
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
},
}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
})

beaconBlockRoot := bytesutil.ToBytes32([]byte("A"))
targetRoot := bytesutil.ToBytes32([]byte("B"))
Expand Down Expand Up @@ -368,14 +360,12 @@ func TestAttestToBlockHead_BlocksDoubleAtt(t *testing.T) {
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
},
}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
})
beaconBlockRoot := bytesutil.ToBytes32([]byte("A"))
targetRoot := bytesutil.ToBytes32([]byte("B"))
sourceRoot := bytesutil.ToBytes32([]byte("C"))
Expand Down Expand Up @@ -424,14 +414,12 @@ func TestAttestToBlockHead_BlocksSurroundAtt(t *testing.T) {
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
},
}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
})
beaconBlockRoot := bytesutil.ToBytes32([]byte("A"))
targetRoot := bytesutil.ToBytes32([]byte("B"))
sourceRoot := bytesutil.ToBytes32([]byte("C"))
Expand Down Expand Up @@ -480,14 +468,12 @@ func TestAttestToBlockHead_BlocksSurroundedAtt(t *testing.T) {
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
},
}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
})
beaconBlockRoot := bytesutil.ToBytes32([]byte("A"))
targetRoot := bytesutil.ToBytes32([]byte("B"))
sourceRoot := bytesutil.ToBytes32([]byte("C"))
Expand Down Expand Up @@ -575,13 +561,12 @@ func TestAttestToBlockHead_DoesAttestAfterDelay(t *testing.T) {
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
}}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
})

m.validatorClient.EXPECT().AttestationData(
gomock.Any(), // ctx
Expand Down Expand Up @@ -618,13 +603,12 @@ func TestAttestToBlockHead_CorrectBitfieldLength(t *testing.T) {
committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
var pubKey [fieldparams.BLSPubkeyLength]byte
copy(pubKey[:], validatorKey.PublicKey().Marshal())
validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
}}}
validator.duties = testDutyStore(&ethpb.ValidatorDuty{
PublicKey: validatorKey.PublicKey().Marshal(),
CommitteeIndex: 5,
CommitteeLength: uint64(len(committee)),
ValidatorIndex: validatorIndex,
})
m.validatorClient.EXPECT().AttestationData(
gomock.Any(), // ctx
gomock.AssignableToTypeOf(&ethpb.AttestationDataRequest{}),
Expand Down
Loading
Loading