attempting to improve duties v2 (#15784)

* attempting to improve duties v2

* removing go routine

* changelog

* unnessesary variable

* fixing test

* small optimization existing early on  CommitteeAssignments function

* fixing small bug

* fixes performance issues with duties v2

* fixed changelog

* gofmt

beacon-chain/core/helpers/beacon_committee.go

@@ -399,7 +399,6 @@ func CommitteeAssignments(ctx context.Context, state state.BeaconState, epoch pr
 	ctx, span := trace.StartSpan(ctx, "helpers.CommitteeAssignments")
 	defer span.End()
 
-	// Verify if the epoch is valid for assignment based on the provided state.
 	if err := VerifyAssignmentEpoch(epoch, state); err != nil {
 		return nil, err
 	}
@@ -407,12 +406,15 @@ func CommitteeAssignments(ctx context.Context, state state.BeaconState, epoch pr
 	if err != nil {
 		return nil, err
 	}
-	vals := make(map[primitives.ValidatorIndex]struct{})
+
+	// Deduplicate and make set for O(1) membership checks.
+	vals := make(map[primitives.ValidatorIndex]struct{}, len(validators))
 	for _, v := range validators {
 		vals[v] = struct{}{}
 	}
-	assignments := make(map[primitives.ValidatorIndex]*CommitteeAssignment)
-	// Compute committee assignments for each slot in the epoch.
+	remaining := len(vals)
+
+	assignments := make(map[primitives.ValidatorIndex]*CommitteeAssignment, len(vals))
 	for slot := startSlot; slot < startSlot+params.BeaconConfig().SlotsPerEpoch; slot++ {
 		committees, err := BeaconCommittees(ctx, state, slot)
 		if err != nil {
@@ -420,7 +422,7 @@ func CommitteeAssignments(ctx context.Context, state state.BeaconState, epoch pr
 		}
 		for j, committee := range committees {
 			for _, vIndex := range committee {
-				if _, ok := vals[vIndex]; !ok { // Skip if the validator is not in the provided validators slice.
+				if _, ok := vals[vIndex]; !ok {
 					continue
 				}
 				if _, ok := assignments[vIndex]; !ok {
@@ -429,6 +431,11 @@ func CommitteeAssignments(ctx context.Context, state state.BeaconState, epoch pr
 				assignments[vIndex].Committee = committee
 				assignments[vIndex].AttesterSlot = slot
 				assignments[vIndex].CommitteeIndex = primitives.CommitteeIndex(j)
+				delete(vals, vIndex)
+				remaining--
+				if remaining == 0 {
+					return assignments, nil // early exit
+				}
 			}
 		}
 	}

beacon-chain/rpc/prysm/v1alpha1/validator/duties_v2.go

@@ -17,13 +17,6 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-const (
-	// validatorLookupThreshold determines when to use full assignment map vs cached linear search.
-	// For requests with fewer validators, we use cached linear search to avoid the overhead
-	// of building a complete assignment map for all validators in the epoch.
-	validatorLookupThreshold = 3000
-)
-
 // GetDutiesV2 returns the duties assigned to a list of validators specified
 // in the request object.
 //
@@ -60,7 +53,26 @@ func (vs *Server) dutiesv2(ctx context.Context, req *ethpb.DutiesRequest) (*ethp
 	span.SetAttributes(trace.Int64Attribute("num_pubkeys", int64(len(req.PublicKeys))))
 	defer span.End()
 
-	meta, err := loadDutiesMetadata(ctx, s, req.Epoch, len(req.PublicKeys))
+	// Collect validator indices from public keys and cache the lookups
+	type validatorInfo struct {
+		index primitives.ValidatorIndex
+		found bool
+	}
+	validatorLookup := make(map[string]validatorInfo, len(req.PublicKeys))
+	requestIndices := make([]primitives.ValidatorIndex, 0, len(req.PublicKeys))
+
+	for _, pubKey := range req.PublicKeys {
+		key := string(pubKey)
+		if _, exists := validatorLookup[key]; !exists {
+			idx, ok := s.ValidatorIndexByPubkey(bytesutil.ToBytes48(pubKey))
+			validatorLookup[key] = validatorInfo{index: idx, found: ok}
+			if ok {
+				requestIndices = append(requestIndices, idx)
+			}
+		}
+	}
+
+	meta, err := loadDutiesMetadata(ctx, s, req.Epoch, requestIndices)
 	if err != nil {
 		return nil, err
 	}
@@ -68,14 +80,14 @@ func (vs *Server) dutiesv2(ctx context.Context, req *ethpb.DutiesRequest) (*ethp
 	validatorAssignments := make([]*ethpb.DutiesV2Response_Duty, 0, len(req.PublicKeys))
 	nextValidatorAssignments := make([]*ethpb.DutiesV2Response_Duty, 0, len(req.PublicKeys))
 
-	// start loop for assignments for current and next epochs
+	// Build duties using cached validator index lookups
 	for _, pubKey := range req.PublicKeys {
 		if ctx.Err() != nil {
 			return nil, status.Errorf(codes.Aborted, "Could not continue fetching assignments: %v", ctx.Err())
 		}
 
-		validatorIndex, ok := s.ValidatorIndexByPubkey(bytesutil.ToBytes48(pubKey))
-		if !ok {
+		info := validatorLookup[string(pubKey)]
+		if !info.found {
 			unknownDuty := &ethpb.DutiesV2Response_Duty{
 				PublicKey: pubKey,
 				Status:    ethpb.ValidatorStatus_UNKNOWN_STATUS,
@@ -85,16 +97,15 @@ func (vs *Server) dutiesv2(ctx context.Context, req *ethpb.DutiesRequest) (*ethp
 			continue
 		}
 
-		meta.current.liteAssignment = vs.getValidatorAssignment(meta.current, validatorIndex)
+		currentAssignment := vs.getValidatorAssignment(meta.current, info.index)
+		nextAssignment := vs.getValidatorAssignment(meta.next, info.index)
 
-		meta.next.liteAssignment = vs.getValidatorAssignment(meta.next, validatorIndex)
-
-		assignment, nextAssignment, err := vs.buildValidatorDuty(pubKey, validatorIndex, s, req.Epoch, meta)
+		assignment, nextDuty, err := vs.buildValidatorDuty(pubKey, info.index, s, req.Epoch, meta, currentAssignment, nextAssignment)
 		if err != nil {
 			return nil, err
 		}
 		validatorAssignments = append(validatorAssignments, assignment)
-		nextValidatorAssignments = append(nextValidatorAssignments, nextAssignment)
+		nextValidatorAssignments = append(nextValidatorAssignments, nextDuty)
 	}
 
 	// Dependent roots for fork choice
@@ -147,18 +158,15 @@ type dutiesMetadata struct {
 }
 
 type metadata struct {
-	committeesAtSlot       uint64
-	proposalSlots          map[primitives.ValidatorIndex][]primitives.Slot
-	startSlot              primitives.Slot
-	committeesBySlot       [][][]primitives.ValidatorIndex
-	validatorAssignmentMap map[primitives.ValidatorIndex]*helpers.LiteAssignment
-	liteAssignment         *helpers.LiteAssignment
+	committeesAtSlot     uint64
+	proposalSlots        map[primitives.ValidatorIndex][]primitives.Slot
+	committeeAssignments map[primitives.ValidatorIndex]*helpers.CommitteeAssignment
 }
 
-func loadDutiesMetadata(ctx context.Context, s state.BeaconState, reqEpoch primitives.Epoch, numValidators int) (*dutiesMetadata, error) {
+func loadDutiesMetadata(ctx context.Context, s state.BeaconState, reqEpoch primitives.Epoch, requestIndices []primitives.ValidatorIndex) (*dutiesMetadata, error) {
 	meta := &dutiesMetadata{}
 	var err error
-	meta.current, err = loadMetadata(ctx, s, reqEpoch, numValidators)
+	meta.current, err = loadMetadata(ctx, s, reqEpoch, requestIndices)
 	if err != nil {
 		return nil, err
 	}
@@ -168,14 +176,14 @@ func loadDutiesMetadata(ctx context.Context, s state.BeaconState, reqEpoch primi
 		return nil, status.Errorf(codes.Internal, "Could not compute proposer slots: %v", err)
 	}
 
-	meta.next, err = loadMetadata(ctx, s, reqEpoch+1, numValidators)
+	meta.next, err = loadMetadata(ctx, s, reqEpoch+1, requestIndices)
 	if err != nil {
 		return nil, err
 	}
 	return meta, nil
 }
 
-func loadMetadata(ctx context.Context, s state.BeaconState, reqEpoch primitives.Epoch, numValidators int) (*metadata, error) {
+func loadMetadata(ctx context.Context, s state.BeaconState, reqEpoch primitives.Epoch, requestIndices []primitives.ValidatorIndex) (*metadata, error) {
 	meta := &metadata{}
 
 	if err := helpers.VerifyAssignmentEpoch(reqEpoch, s); err != nil {
@@ -188,56 +196,36 @@ func loadMetadata(ctx context.Context, s state.BeaconState, reqEpoch primitives.
 	}
 	meta.committeesAtSlot = helpers.SlotCommitteeCount(activeValidatorCount)
 
-	meta.startSlot, err = slots.EpochStart(reqEpoch)
+	// Use CommitteeAssignments which only computes committees for requested validators
+	meta.committeeAssignments, err = helpers.CommitteeAssignments(ctx, s, reqEpoch, requestIndices)
 	if err != nil {
-		return nil, err
-	}
-
-	meta.committeesBySlot, err = helpers.PrecomputeCommittees(ctx, s, meta.startSlot)
-	if err != nil {
-		return nil, err
-	}
-
-	if numValidators >= validatorLookupThreshold {
-		meta.validatorAssignmentMap = buildValidatorAssignmentMap(meta.committeesBySlot, meta.startSlot)
+		return nil, status.Errorf(codes.Internal, "Could not compute committee assignments: %v", err)
 	}
 
 	return meta, nil
 }
 
-// buildValidatorAssignmentMap creates a map from validator index to assignment for O(1) lookup.
-func buildValidatorAssignmentMap(
-	bySlot [][][]primitives.ValidatorIndex,
-	startSlot primitives.Slot,
-) map[primitives.ValidatorIndex]*helpers.LiteAssignment {
-	validatorToAssignment := make(map[primitives.ValidatorIndex]*helpers.LiteAssignment)
-
-	for relativeSlot, committees := range bySlot {
-		for cIdx, committee := range committees {
-			for pos, vIdx := range committee {
-				validatorToAssignment[vIdx] = &helpers.LiteAssignment{
-					AttesterSlot:            startSlot + primitives.Slot(relativeSlot),
-					CommitteeIndex:          primitives.CommitteeIndex(cIdx),
-					CommitteeLength:         uint64(len(committee)),
-					ValidatorCommitteeIndex: uint64(pos),
-				}
-			}
+// findValidatorIndexInCommittee finds the position of a validator in a committee.
+func findValidatorIndexInCommittee(committee []primitives.ValidatorIndex, validatorIndex primitives.ValidatorIndex) uint64 {
+	for i, vIdx := range committee {
+		if vIdx == validatorIndex {
+			return uint64(i)
 		}
 	}
-	return validatorToAssignment
+	return 0
 }
 
-// getValidatorAssignment retrieves the assignment for a validator using either
-// the pre-built assignment map (for large requests) or linear search (for small requests).
+// getValidatorAssignment retrieves the assignment for a validator from CommitteeAssignments.
 func (vs *Server) getValidatorAssignment(meta *metadata, validatorIndex primitives.ValidatorIndex) *helpers.LiteAssignment {
-	if meta.validatorAssignmentMap != nil {
-		if assignment, exists := meta.validatorAssignmentMap[validatorIndex]; exists {
-			return assignment
+	if assignment, exists := meta.committeeAssignments[validatorIndex]; exists {
+		return &helpers.LiteAssignment{
+			AttesterSlot:            assignment.AttesterSlot,
+			CommitteeIndex:          assignment.CommitteeIndex,
+			CommitteeLength:         uint64(len(assignment.Committee)),
+			ValidatorCommitteeIndex: findValidatorIndexInCommittee(assignment.Committee, validatorIndex),
 		}
-		return &helpers.LiteAssignment{}
 	}
-
-	return helpers.AssignmentForValidator(meta.committeesBySlot, meta.startSlot, validatorIndex)
+	return &helpers.LiteAssignment{}
 }
 
 // buildValidatorDuty builds both current‑epoch and next‑epoch V2 duty objects
@@ -248,21 +236,23 @@ func (vs *Server) buildValidatorDuty(
 	s state.BeaconState,
 	reqEpoch primitives.Epoch,
 	meta *dutiesMetadata,
+	currentAssignment *helpers.LiteAssignment,
+	nextAssignment *helpers.LiteAssignment,
 ) (*ethpb.DutiesV2Response_Duty, *ethpb.DutiesV2Response_Duty, error) {
 	assignment := &ethpb.DutiesV2Response_Duty{PublicKey: pubKey}
-	nextAssignment := &ethpb.DutiesV2Response_Duty{PublicKey: pubKey}
+	nextDuty := &ethpb.DutiesV2Response_Duty{PublicKey: pubKey}
 
 	statusEnum := assignmentStatus(s, idx)
 	assignment.ValidatorIndex = idx
 	assignment.Status = statusEnum
 	assignment.CommitteesAtSlot = meta.current.committeesAtSlot
 	assignment.ProposerSlots = meta.current.proposalSlots[idx]
-	populateCommitteeFields(assignment, meta.current.liteAssignment)
+	populateCommitteeFields(assignment, currentAssignment)
 
-	nextAssignment.ValidatorIndex = idx
-	nextAssignment.Status = statusEnum
-	nextAssignment.CommitteesAtSlot = meta.next.committeesAtSlot
-	populateCommitteeFields(nextAssignment, meta.next.liteAssignment)
+	nextDuty.ValidatorIndex = idx
+	nextDuty.Status = statusEnum
+	nextDuty.CommitteesAtSlot = meta.next.committeesAtSlot
+	populateCommitteeFields(nextDuty, nextAssignment)
 
 	// Sync committee flags
 	if coreTime.HigherEqualThanAltairVersionAndEpoch(s, reqEpoch) {
@@ -271,7 +261,7 @@ func (vs *Server) buildValidatorDuty(
 			return nil, nil, status.Errorf(codes.Internal, "Could not determine current epoch sync committee: %v", err)
 		}
 		assignment.IsSyncCommittee = inSync
-		nextAssignment.IsSyncCommittee = inSync
+		nextDuty.IsSyncCommittee = inSync
 		if inSync {
 			if err := core.RegisterSyncSubnetCurrentPeriodProto(s, reqEpoch, pubKey, statusEnum); err != nil {
 				return nil, nil, status.Errorf(codes.Internal, "Could not register sync subnet current period: %v", err)
@@ -290,18 +280,16 @@ func (vs *Server) buildValidatorDuty(
 			if err != nil {
 				return nil, nil, status.Errorf(codes.Internal, "Could not determine next epoch sync committee: %v", err)
 			}
-			nextAssignment.IsSyncCommittee = nextInSync
+			nextDuty.IsSyncCommittee = nextInSync
 			if nextInSync {
-				go func() {
-					if err := core.RegisterSyncSubnetNextPeriodProto(s, reqEpoch, pubKey, statusEnum); err != nil {
-						log.WithError(err).Warn("Could not register sync subnet next period")
-					}
-				}()
+				if err := core.RegisterSyncSubnetNextPeriodProto(s, reqEpoch, pubKey, statusEnum); err != nil {
+					log.WithError(err).Warn("Could not register sync subnet next period")
+				}
 			}
 		}
 	}
 
-	return assignment, nextAssignment, nil
+	return assignment, nextDuty, nil
 }
 
 func populateCommitteeFields(duty *ethpb.DutiesV2Response_Duty, la *helpers.LiteAssignment) {

beacon-chain/rpc/prysm/v1alpha1/validator/duties_v2_test.go

@@ -560,105 +560,20 @@ func TestGetDutiesV2_SyncNotReady(t *testing.T) {
 	assert.ErrorContains(t, "Syncing to latest head", err)
 }
 
-func TestBuildValidatorAssignmentMap(t *testing.T) {
-	start := primitives.Slot(200)
-	bySlot := [][][]primitives.ValidatorIndex{
-		{{1, 2, 3}},        // slot 200, committee 0
-		{{7, 8, 9}},        // slot 201, committee 0
-		{{4, 5}, {10, 11}}, // slot 202, committee 0 & 1
-	}
-
-	assignmentMap := buildValidatorAssignmentMap(bySlot, start)
-
-	// Test validator 8 assignment (slot 201, committee 0, position 1)
-	vIdx := primitives.ValidatorIndex(8)
-	got, exists := assignmentMap[vIdx]
-	assert.Equal(t, true, exists)
-	require.NotNil(t, got)
-	assert.Equal(t, start+1, got.AttesterSlot)
-	assert.Equal(t, primitives.CommitteeIndex(0), got.CommitteeIndex)
-	assert.Equal(t, uint64(3), got.CommitteeLength)
-	assert.Equal(t, uint64(1), got.ValidatorCommitteeIndex)
-
-	// Test validator 1 assignment (slot 200, committee 0, position 0)
-	vIdx1 := primitives.ValidatorIndex(1)
-	got1, exists1 := assignmentMap[vIdx1]
-	assert.Equal(t, true, exists1)
-	require.NotNil(t, got1)
-	assert.Equal(t, start, got1.AttesterSlot)
-	assert.Equal(t, primitives.CommitteeIndex(0), got1.CommitteeIndex)
-	assert.Equal(t, uint64(3), got1.CommitteeLength)
-	assert.Equal(t, uint64(0), got1.ValidatorCommitteeIndex)
-
-	// Test validator 10 assignment (slot 202, committee 1, position 0)
-	vIdx10 := primitives.ValidatorIndex(10)
-	got10, exists10 := assignmentMap[vIdx10]
-	assert.Equal(t, true, exists10)
-	require.NotNil(t, got10)
-	assert.Equal(t, start+2, got10.AttesterSlot)
-	assert.Equal(t, primitives.CommitteeIndex(1), got10.CommitteeIndex)
-	assert.Equal(t, uint64(2), got10.CommitteeLength)
-	assert.Equal(t, uint64(0), got10.ValidatorCommitteeIndex)
-
-	// Test non-existent validator
-	_, exists99 := assignmentMap[primitives.ValidatorIndex(99)]
-	assert.Equal(t, false, exists99)
-
-	// Verify that we get the same results as the linear search
-	for _, committees := range bySlot {
-		for _, committee := range committees {
-			for _, validatorIdx := range committee {
-				linearResult := helpers.AssignmentForValidator(bySlot, start, validatorIdx)
-				mapResult, mapExists := assignmentMap[validatorIdx]
-				assert.Equal(t, true, mapExists)
-				require.DeepEqual(t, linearResult, mapResult)
-			}
-		}
-	}
-}
-
-func TestGetValidatorAssignment_WithAssignmentMap(t *testing.T) {
+func TestGetValidatorAssignment(t *testing.T) {
 	start := primitives.Slot(100)
-	bySlot := [][][]primitives.ValidatorIndex{
-		{{1, 2, 3}},
-		{{4, 5, 6}},
+
+	// Test using CommitteeAssignments
+	committeeAssignments := map[primitives.ValidatorIndex]*helpers.CommitteeAssignment{
+		5: {
+			Committee:      []primitives.ValidatorIndex{4, 5, 6},
+			AttesterSlot:   start + 1,
+			CommitteeIndex: primitives.CommitteeIndex(0),
+		},
 	}
 
-	// Test with pre-built assignment map (large request scenario)
 	meta := &metadata{
-		startSlot:              start,
-		committeesBySlot:       bySlot,
-		validatorAssignmentMap: buildValidatorAssignmentMap(bySlot, start),
-	}
-
-	vs := &Server{}
-
-	// Test existing validator (validator 2 is at position 1 in the committee, not position 2)
-	assignment := vs.getValidatorAssignment(meta, primitives.ValidatorIndex(2))
-	require.NotNil(t, assignment)
-	assert.Equal(t, start, assignment.AttesterSlot)
-	assert.Equal(t, primitives.CommitteeIndex(0), assignment.CommitteeIndex)
-	assert.Equal(t, uint64(1), assignment.ValidatorCommitteeIndex)
-
-	// Test non-existent validator should return empty assignment
-	assignment = vs.getValidatorAssignment(meta, primitives.ValidatorIndex(99))
-	require.NotNil(t, assignment)
-	assert.Equal(t, primitives.Slot(0), assignment.AttesterSlot)
-	assert.Equal(t, primitives.CommitteeIndex(0), assignment.CommitteeIndex)
-}
-
-func TestGetValidatorAssignment_WithoutAssignmentMap(t *testing.T) {
-	start := primitives.Slot(100)
-	bySlot := [][][]primitives.ValidatorIndex{
-		{{1, 2, 3}},
-		{{4, 5, 6}},
-	}
-
-	// Test without assignment map (small request scenario)
-	meta := &metadata{
-		startSlot:              start,
-		committeesBySlot:       bySlot,
-		validatorAssignmentMap: nil, // No map - should use linear search
+		committeeAssignments: committeeAssignments,
 	}
 
 	vs := &Server{}
@@ -676,53 +591,3 @@ func TestGetValidatorAssignment_WithoutAssignmentMap(t *testing.T) {
 	assert.Equal(t, primitives.Slot(0), assignment.AttesterSlot)
 	assert.Equal(t, primitives.CommitteeIndex(0), assignment.CommitteeIndex)
 }
-
-func TestLoadMetadata_ThresholdBehavior(t *testing.T) {
-	state, _ := util.DeterministicGenesisState(t, 128)
-	epoch := primitives.Epoch(0)
-
-	tests := []struct {
-		name                string
-		numValidators       int
-		expectAssignmentMap bool
-	}{
-		{
-			name:                "Small request - below threshold",
-			numValidators:       100,
-			expectAssignmentMap: false,
-		},
-		{
-			name:                "Large request - at threshold",
-			numValidators:       validatorLookupThreshold,
-			expectAssignmentMap: true,
-		},
-		{
-			name:                "Large request - above threshold",
-			numValidators:       validatorLookupThreshold + 1000,
-			expectAssignmentMap: true,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			meta, err := loadMetadata(t.Context(), state, epoch, tt.numValidators)
-			require.NoError(t, err)
-			require.NotNil(t, meta)
-
-			if tt.expectAssignmentMap {
-				require.NotNil(t, meta.validatorAssignmentMap, "Expected assignment map to be built for large requests")
-				assert.Equal(t, true, len(meta.validatorAssignmentMap) > 0, "Assignment map should not be empty")
-			} else {
-				// For small requests, the map should be nil (not initialized)
-				if meta.validatorAssignmentMap != nil {
-					t.Errorf("Expected no assignment map for small requests, got: %v", meta.validatorAssignmentMap)
-				}
-			}
-
-			// Common fields should always be set
-			assert.Equal(t, true, meta.committeesAtSlot > 0)
-			require.NotNil(t, meta.committeesBySlot)
-			assert.Equal(t, true, len(meta.committeesBySlot) > 0)
-		})
-	}
-}

changelog/james-prysm_improve-duties-v2.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- adding in improvements to getduties v2, replaces helpers.PrecomputeCommittees() ( exepensive ) with CommitteeAssignments