Implement ListIndexedAttestations Endpoint in Prysm (#4892)

* update patch and workspace

* stub methods

* implementation of indexed attestations list

* include latest ethereumapis

* update request type

* compute committee pure function

* use compute committee helper

* add test into list indexed attestations

* regenerate mock

* imports and out of range check

* test passing for archived epoch

* add comment

* comment

* better comment on func

* throw in continue instead

WORKSPACE

@@ -1272,7 +1272,7 @@ go_repository(
 
 go_repository(
     name = "com_github_prysmaticlabs_ethereumapis",
-    commit = "b7452dde4ca361809def4ed5924ab3cb7ad1299a",
+    commit = "53ccc146f7f488c5c7634530057f4aedf510a9ac",
     importpath = "github.com/prysmaticlabs/ethereumapis",
     patch_args = ["-p1"],
     patches = [

beacon-chain/rpc/beacon/BUILD.bazel

@@ -30,6 +30,7 @@ go_library(
         "//beacon-chain/operations/slashings:go_default_library",
         "//beacon-chain/powchain:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
+        "//shared/attestationutil:go_default_library",
         "//shared/bytesutil:go_default_library",
         "//shared/hashutil:go_default_library",
         "//shared/pagination:go_default_library",
@@ -73,6 +74,7 @@ go_test(
         "//beacon-chain/rpc/testing:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
+        "//shared/attestationutil:go_default_library",
         "//shared/params:go_default_library",
         "//shared/slotutil/testing:go_default_library",
         "@com_github_gogo_protobuf//proto:go_default_library",

beacon-chain/rpc/beacon/attestations.go

@@ -8,9 +8,12 @@ import (
 	ptypes "github.com/gogo/protobuf/types"
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
 	"github.com/prysmaticlabs/go-ssz"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db/filters"
 	"github.com/prysmaticlabs/prysm/beacon-chain/flags"
+	"github.com/prysmaticlabs/prysm/shared/attestationutil"
 	"github.com/prysmaticlabs/prysm/shared/pagination"
+	"github.com/prysmaticlabs/prysm/shared/params"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -110,6 +113,97 @@ func (bs *Server) ListAttestations(
 	}, nil
 }
 
+// ListIndexedAttestations retrieves indexed attestations by target epoch.
+// IndexedAttestations are sorted by data slot by default. Either a target epoch filter
+// or a boolean filter specifying a request for genesis epoch attestations may be used.
+//
+// The server may return an empty list when no attestations match the given
+// filter criteria. This RPC should not return NOT_FOUND. Only one filter
+// criteria should be used.
+func (bs *Server) ListIndexedAttestations(
+	ctx context.Context, req *ethpb.ListIndexedAttestationsRequest,
+) (*ethpb.ListIndexedAttestationsResponse, error) {
+	atts := make([]*ethpb.Attestation, 0)
+	var err error
+	epoch := helpers.SlotToEpoch(bs.GenesisTimeFetcher.CurrentSlot())
+	switch q := req.QueryFilter.(type) {
+	case *ethpb.ListIndexedAttestationsRequest_TargetEpoch:
+		atts, err = bs.BeaconDB.Attestations(ctx, filters.NewFilter().SetTargetEpoch(q.TargetEpoch))
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "Could not fetch attestations: %v", err)
+		}
+		epoch = q.TargetEpoch
+	case *ethpb.ListIndexedAttestationsRequest_GenesisEpoch:
+		atts, err = bs.BeaconDB.Attestations(ctx, filters.NewFilter().SetTargetEpoch(0))
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "Could not fetch attestations: %v", err)
+		}
+		epoch = 0
+	default:
+		return nil, status.Error(codes.InvalidArgument, "Must specify a filter criteria for fetching attestations")
+	}
+	// We sort attestations according to the Sortable interface.
+	sort.Sort(sortableAttestations(atts))
+	numAttestations := len(atts)
+
+	// If there are no attestations, we simply return a response specifying this.
+	// Otherwise, attempting to paginate 0 attestations below would result in an error.
+	if numAttestations == 0 {
+		return &ethpb.ListIndexedAttestationsResponse{
+			IndexedAttestations: make([]*ethpb.IndexedAttestation, 0),
+			TotalSize:           int32(0),
+			NextPageToken:       strconv.Itoa(0),
+		}, nil
+	}
+
+	committeesBySlot, _, err := bs.retrieveCommitteesForEpoch(ctx, epoch)
+	if err != nil {
+		return nil, status.Errorf(
+			codes.Internal,
+			"Could not retrieve committees for epoch %d: %v",
+			epoch,
+			err,
+		)
+	}
+
+	// We use the retrieved committees for the epoch to convert all attestations
+	// into indexed form effectively.
+	indexedAtts := make([]*ethpb.IndexedAttestation, numAttestations, numAttestations)
+	startSlot := helpers.StartSlot(epoch)
+	endSlot := startSlot + params.BeaconConfig().SlotsPerEpoch
+	for i := 0; i < len(indexedAtts); i++ {
+		att := atts[i]
+		// Out of range check, the attestation slot cannot be greater
+		// the last slot of the requested epoch or smaller than its start slot
+		// given committees are accessed as a map of slot -> commitees list, where there are
+		// SLOTS_PER_EPOCH keys in the map.
+		if att.Data.Slot < startSlot || att.Data.Slot > endSlot {
+			continue
+		}
+		committee := committeesBySlot[att.Data.Slot].Committees[att.Data.CommitteeIndex]
+		idxAtt, err := attestationutil.ConvertToIndexed(ctx, atts[i], committee.ValidatorIndices)
+		if err != nil {
+			return nil, status.Errorf(
+				codes.Internal,
+				"Could not convert attestation with slot %d to indexed form: %v",
+				att.Data.Slot,
+				err,
+			)
+		}
+		indexedAtts[i] = idxAtt
+	}
+
+	start, end, nextPageToken, err := pagination.StartAndEndPage(req.PageToken, int(req.PageSize), len(indexedAtts))
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Could not paginate attestations: %v", err)
+	}
+	return &ethpb.ListIndexedAttestationsResponse{
+		IndexedAttestations: indexedAtts[start:end],
+		TotalSize:           int32(len(indexedAtts)),
+		NextPageToken:       nextPageToken,
+	}, nil
+}
+
 // StreamAttestations to clients at the end of every slot. This method retrieves the
 // aggregated attestations currently in the pool at the start of a slot and sends
 // them over a gRPC stream.
@@ -133,6 +227,15 @@ func (bs *Server) StreamAttestations(
 	}
 }
 
+// StreamIndexedAttestations to clients at the end of every slot. This method retrieves the
+// aggregated attestations currently in the pool, converts them into indexed form, and
+// sends them over a gRPC stream.
+func (bs *Server) StreamIndexedAttestations(
+	_ *ptypes.Empty, stream ethpb.BeaconChain_StreamIndexedAttestationsServer,
+) error {
+	return status.Error(codes.Unimplemented, "Unimplemented")
+}
+
 // AttestationPool retrieves pending attestations.
 //
 // The server returns a list of attestations that have been seen but not

beacon-chain/rpc/beacon/attestations_test.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/gogo/protobuf/proto"
 	ptypes "github.com/gogo/protobuf/types"
@@ -15,12 +16,14 @@ import (
 	"github.com/prysmaticlabs/go-bitfield"
 	"github.com/prysmaticlabs/go-ssz"
 	mock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
 	dbTest "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
 	"github.com/prysmaticlabs/prysm/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/beacon-chain/operations/attestations"
 	mockRPC "github.com/prysmaticlabs/prysm/beacon-chain/rpc/testing"
 	stateTrie "github.com/prysmaticlabs/prysm/beacon-chain/state"
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	"github.com/prysmaticlabs/prysm/shared/attestationutil"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	mocktick "github.com/prysmaticlabs/prysm/shared/slotutil/testing"
 )
@@ -536,6 +539,196 @@ func TestServer_ListAttestations_Pagination_DefaultPageSize(t *testing.T) {
 	}
 }
 
+func TestServer_ListIndexedAttestations_GenesisEpoch(t *testing.T) {
+	db := dbTest.SetupDB(t)
+	defer dbTest.TeardownDB(t, db)
+	helpers.ClearCache()
+	ctx := context.Background()
+
+	count := params.BeaconConfig().SlotsPerEpoch
+	atts := make([]*ethpb.Attestation, 0, count)
+	for i := uint64(0); i < count; i++ {
+		attExample := &ethpb.Attestation{
+			Data: &ethpb.AttestationData{
+				BeaconBlockRoot: []byte("root"),
+				Slot:            i,
+				CommitteeIndex:  0,
+				Target: &ethpb.Checkpoint{
+					Epoch: 0,
+					Root:  make([]byte, 32),
+				},
+			},
+			AggregationBits: bitfield.Bitlist{0b11},
+		}
+		atts = append(atts, attExample)
+	}
+	if err := db.SaveAttestations(ctx, atts); err != nil {
+		t.Fatal(err)
+	}
+
+	// We setup 128 validators.
+	numValidators := 128
+	headState := setupActiveValidators(t, db, numValidators)
+
+	randaoMixes := make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector)
+	for i := 0; i < len(randaoMixes); i++ {
+		randaoMixes[i] = make([]byte, 32)
+	}
+	if err := headState.SetRandaoMixes(randaoMixes); err != nil {
+		t.Fatal(err)
+	}
+
+	activeIndices, err := helpers.ActiveValidatorIndices(headState, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	epoch := uint64(0)
+	attesterSeed, err := helpers.Seed(headState, epoch, params.BeaconConfig().DomainBeaconAttester)
+	if err != nil {
+		t.Fatal(err)
+	}
+	committees, err := computeCommittees(helpers.StartSlot(epoch), activeIndices, attesterSeed)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Next up we convert the test attestations to indexed form:
+	indexedAtts := make([]*ethpb.IndexedAttestation, len(atts), len(atts))
+	for i := 0; i < len(indexedAtts); i++ {
+		att := atts[i]
+		committee := committees[att.Data.Slot].Committees[att.Data.CommitteeIndex]
+		idxAtt, err := attestationutil.ConvertToIndexed(ctx, atts[i], committee.ValidatorIndices)
+		if err != nil {
+			t.Fatalf("Could not convert attestation to indexed: %v", err)
+		}
+		indexedAtts[i] = idxAtt
+	}
+
+	bs := &Server{
+		BeaconDB: db,
+		HeadFetcher: &mock.ChainService{
+			State: headState,
+		},
+		GenesisTimeFetcher: &mock.ChainService{
+			Genesis: time.Now(),
+		},
+	}
+
+	res, err := bs.ListIndexedAttestations(ctx, &ethpb.ListIndexedAttestationsRequest{
+		QueryFilter: &ethpb.ListIndexedAttestationsRequest_GenesisEpoch{
+			GenesisEpoch: true,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(indexedAtts, res.IndexedAttestations) {
+		t.Fatalf(
+			"Incorrect list indexed attestations response: wanted %v, received %v",
+			indexedAtts,
+			res.IndexedAttestations,
+		)
+	}
+}
+
+func TestServer_ListIndexedAttestations_ArchivedEpoch(t *testing.T) {
+	db := dbTest.SetupDB(t)
+	defer dbTest.TeardownDB(t, db)
+	helpers.ClearCache()
+	ctx := context.Background()
+
+	count := params.BeaconConfig().SlotsPerEpoch
+	atts := make([]*ethpb.Attestation, 0, count)
+	startSlot := helpers.StartSlot(50)
+	epoch := uint64(50)
+	for i := startSlot; i < count; i++ {
+		attExample := &ethpb.Attestation{
+			Data: &ethpb.AttestationData{
+				BeaconBlockRoot: []byte("root"),
+				Slot:            i,
+				CommitteeIndex:  0,
+				Target: &ethpb.Checkpoint{
+					Epoch: epoch,
+					Root:  make([]byte, 32),
+				},
+			},
+			AggregationBits: bitfield.Bitlist{0b11},
+		}
+		atts = append(atts, attExample)
+	}
+	if err := db.SaveAttestations(ctx, atts); err != nil {
+		t.Fatal(err)
+	}
+
+	// We setup 128 validators.
+	numValidators := 128
+	headState := setupActiveValidators(t, db, numValidators)
+
+	randaoMixes := make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector)
+	for i := 0; i < len(randaoMixes); i++ {
+		randaoMixes[i] = make([]byte, 32)
+	}
+	if err := headState.SetRandaoMixes(randaoMixes); err != nil {
+		t.Fatal(err)
+	}
+	if err := headState.SetSlot(startSlot); err != nil {
+		t.Fatal(err)
+	}
+
+	activeIndices, err := helpers.ActiveValidatorIndices(headState, epoch)
+	if err != nil {
+		t.Fatal(err)
+	}
+	attesterSeed, err := helpers.Seed(headState, epoch, params.BeaconConfig().DomainBeaconAttester)
+	if err != nil {
+		t.Fatal(err)
+	}
+	committees, err := computeCommittees(epoch, activeIndices, attesterSeed)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Next up we convert the test attestations to indexed form:
+	indexedAtts := make([]*ethpb.IndexedAttestation, len(atts), len(atts))
+	for i := 0; i < len(indexedAtts); i++ {
+		att := atts[i]
+		committee := committees[att.Data.Slot].Committees[att.Data.CommitteeIndex]
+		idxAtt, err := attestationutil.ConvertToIndexed(ctx, atts[i], committee.ValidatorIndices)
+		if err != nil {
+			t.Fatalf("Could not convert attestation to indexed: %v", err)
+		}
+		indexedAtts[i] = idxAtt
+	}
+
+	bs := &Server{
+		BeaconDB: db,
+		HeadFetcher: &mock.ChainService{
+			State: headState,
+		},
+		GenesisTimeFetcher: &mock.ChainService{
+			Genesis: time.Now(),
+		},
+	}
+
+	res, err := bs.ListIndexedAttestations(ctx, &ethpb.ListIndexedAttestationsRequest{
+		QueryFilter: &ethpb.ListIndexedAttestationsRequest_TargetEpoch{
+			TargetEpoch: epoch,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(indexedAtts, res.IndexedAttestations) {
+		t.Fatalf(
+			"Incorrect list indexed attestations response: wanted %v, received %v",
+			indexedAtts,
+			res.IndexedAttestations,
+		)
+	}
+}
+
 func TestServer_AttestationPool_Pagination_ExceedsMaxPageSize(t *testing.T) {
 	ctx := context.Background()
 	bs := &Server{}

beacon-chain/rpc/beacon/committees.go

@@ -34,17 +34,35 @@ func (bs *Server) ListBeaconCommittees(
 	default:
 		startSlot = headSlot
 	}
+	committees, activeIndices, err := bs.retrieveCommitteesForEpoch(ctx, helpers.SlotToEpoch(startSlot))
+	if err != nil {
+		return nil, status.Errorf(
+			codes.Internal,
+			"Could not retrieve committees for epoch %d: %v",
+			helpers.SlotToEpoch(startSlot),
+			err,
+		)
+	}
+	return &ethpb.BeaconCommittees{
+		Epoch:                helpers.SlotToEpoch(startSlot),
+		Committees:           committees,
+		ActiveValidatorCount: uint64(len(activeIndices)),
+	}, nil
+}
 
+func (bs *Server) retrieveCommitteesForEpoch(
+	ctx context.Context,
+	epoch uint64,
+) (map[uint64]*ethpb.BeaconCommittees_CommitteesList, []uint64, error) {
 	var attesterSeed [32]byte
 	var activeIndices []uint64
 	var err error
-	// This is the archival condition, if the requested epoch is < previous epoch.
-	headEpoch := helpers.SlotToEpoch(headSlot)
-	// Adding 1 here to prevent underflow on headEpoch.
+	startSlot := helpers.StartSlot(epoch)
+	headEpoch := helpers.SlotToEpoch(bs.HeadFetcher.HeadSlot())
 	if helpers.SlotToEpoch(startSlot)+1 < headEpoch {
 		activeIndices, err = bs.HeadFetcher.HeadValidatorsIndices(helpers.SlotToEpoch(startSlot))
 		if err != nil {
-			return nil, status.Errorf(
+			return nil, nil, status.Errorf(
 				codes.Internal,
 				"Could not retrieve active indices for epoch %d: %v",
 				helpers.SlotToEpoch(startSlot),
@@ -53,7 +71,7 @@ func (bs *Server) ListBeaconCommittees(
 		}
 		archivedCommitteeInfo, err := bs.BeaconDB.ArchivedCommitteeInfo(ctx, helpers.SlotToEpoch(startSlot))
 		if err != nil {
-			return nil, status.Errorf(
+			return nil, nil, status.Errorf(
 				codes.Internal,
 				"Could not request archival data for epoch %d: %v",
 				helpers.SlotToEpoch(startSlot),
@@ -61,7 +79,7 @@ func (bs *Server) ListBeaconCommittees(
 			)
 		}
 		if archivedCommitteeInfo == nil {
-			return nil, status.Errorf(
+			return nil, nil, status.Errorf(
 				codes.NotFound,
 				"Could not retrieve data for epoch %d, perhaps --archive in the running beacon node is disabled",
 				helpers.SlotToEpoch(startSlot),
@@ -73,7 +91,7 @@ func (bs *Server) ListBeaconCommittees(
 		requestedEpoch := helpers.SlotToEpoch(startSlot)
 		activeIndices, err = bs.HeadFetcher.HeadValidatorsIndices(requestedEpoch)
 		if err != nil {
-			return nil, status.Errorf(
+			return nil, nil, status.Errorf(
 				codes.Internal,
 				"Could not retrieve active indices for requested epoch %d: %v",
 				requestedEpoch,
@@ -82,7 +100,7 @@ func (bs *Server) ListBeaconCommittees(
 		}
 		attesterSeed, err = bs.HeadFetcher.HeadSeed(requestedEpoch)
 		if err != nil {
-			return nil, status.Errorf(
+			return nil, nil, status.Errorf(
 				codes.Internal,
 				"Could not retrieve attester seed for requested epoch %d: %v",
 				requestedEpoch,
@@ -91,15 +109,34 @@ func (bs *Server) ListBeaconCommittees(
 		}
 	} else {
 		// Otherwise, we are requesting data from the future and we return an error.
-		return nil, status.Errorf(
+		return nil, nil, status.Errorf(
 			codes.InvalidArgument,
 			"Cannot retrieve information about an epoch in the future, current epoch %d, requesting %d",
-			helpers.SlotToEpoch(headSlot),
+			headEpoch,
 			helpers.SlotToEpoch(startSlot),
 		)
 	}
 
-	committeesList := make(map[uint64]*ethpb.BeaconCommittees_CommitteesList)
+	committeesListsBySlot, err := computeCommittees(startSlot, activeIndices, attesterSeed)
+	if err != nil {
+		return nil, nil, status.Errorf(
+			codes.InvalidArgument,
+			"Could not compute committees for epoch %d: %v",
+			helpers.SlotToEpoch(startSlot),
+			err,
+		)
+	}
+	return committeesListsBySlot, activeIndices, nil
+}
+
+// Compute committees given a start slot, active validator indices, and
+// the attester seeds value.
+func computeCommittees(
+	startSlot uint64,
+	activeIndices []uint64,
+	attesterSeed [32]byte,
+) (map[uint64]*ethpb.BeaconCommittees_CommitteesList, error) {
+	committeesListsBySlot := make(map[uint64]*ethpb.BeaconCommittees_CommitteesList)
 	for slot := startSlot; slot < startSlot+params.BeaconConfig().SlotsPerEpoch; slot++ {
 		var countAtSlot = uint64(len(activeIndices)) / params.BeaconConfig().SlotsPerEpoch / params.BeaconConfig().TargetCommitteeSize
 		if countAtSlot > params.BeaconConfig().MaxCommitteesPerSlot {
@@ -123,14 +160,9 @@ func (bs *Server) ListBeaconCommittees(
 				ValidatorIndices: committee,
 			}
 		}
-		committeesList[slot] = &ethpb.BeaconCommittees_CommitteesList{
+		committeesListsBySlot[slot] = &ethpb.BeaconCommittees_CommitteesList{
 			Committees: committeeItems,
 		}
 	}
-
-	return &ethpb.BeaconCommittees{
-		Epoch:                helpers.SlotToEpoch(startSlot),
-		Committees:           committeesList,
-		ActiveValidatorCount: uint64(len(activeIndices)),
-	}, nil
+	return committeesListsBySlot, nil
 }

beacon-chain/rpc/beacon/committees_test.go

@@ -48,53 +48,24 @@ func TestServer_ListBeaconCommittees_CurrentEpoch(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	wanted := make(map[uint64]*ethpb.BeaconCommittees_CommitteesList)
-	for slot := uint64(0); slot < params.BeaconConfig().SlotsPerEpoch; slot++ {
-		var countAtSlot = uint64(numValidators) / params.BeaconConfig().SlotsPerEpoch / params.BeaconConfig().TargetCommitteeSize
-		if countAtSlot > params.BeaconConfig().MaxCommitteesPerSlot {
-			countAtSlot = params.BeaconConfig().MaxCommitteesPerSlot
-		}
-		if countAtSlot == 0 {
-			countAtSlot = 1
-		}
-		committeeItems := make([]*ethpb.BeaconCommittees_CommitteeItem, countAtSlot)
-		for i := uint64(0); i < countAtSlot; i++ {
-			committee, err := helpers.BeaconCommittee(activeIndices, attesterSeed, slot, i)
-			if err != nil {
-				t.Fatal(err)
-			}
-			committeeItems[i] = &ethpb.BeaconCommittees_CommitteeItem{
-				ValidatorIndices: committee,
-			}
-		}
-		wanted[slot] = &ethpb.BeaconCommittees_CommitteesList{
-			Committees: committeeItems,
-		}
+	committees, err := computeCommittees(0, activeIndices, attesterSeed)
+	if err != nil {
+		t.Fatal(err)
 	}
 
-	tests := []struct {
-		req *ethpb.ListCommitteesRequest
-		res *ethpb.BeaconCommittees
-	}{
-		{
-			req: &ethpb.ListCommitteesRequest{
-				QueryFilter: &ethpb.ListCommitteesRequest_Genesis{Genesis: true},
-			},
-			res: &ethpb.BeaconCommittees{
-				Epoch:                0,
-				Committees:           wanted,
-				ActiveValidatorCount: uint64(numValidators),
-			},
-		},
+	wanted := &ethpb.BeaconCommittees{
+		Epoch:                0,
+		Committees:           committees,
+		ActiveValidatorCount: uint64(numValidators),
 	}
-	for _, test := range tests {
-		res, err := bs.ListBeaconCommittees(context.Background(), test.req)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if !proto.Equal(res, test.res) {
-			t.Errorf("Expected %v, received %v", test.res, res)
-		}
+	res, err := bs.ListBeaconCommittees(context.Background(), &ethpb.ListCommitteesRequest{
+		QueryFilter: &ethpb.ListCommitteesRequest_Genesis{Genesis: true},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !proto.Equal(res, wanted) {
+		t.Errorf("Expected %v, received %v", wanted, res)
 	}
 }
 
@@ -127,31 +98,10 @@ func TestServer_ListBeaconCommittees_PreviousEpoch(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	wanted := make(map[uint64]*ethpb.BeaconCommittees_CommitteesList)
 	startSlot := helpers.StartSlot(1)
-	for slot := startSlot; slot < startSlot+params.BeaconConfig().SlotsPerEpoch; slot++ {
-		var countAtSlot = uint64(numValidators) / params.BeaconConfig().SlotsPerEpoch / params.BeaconConfig().TargetCommitteeSize
-		if countAtSlot > params.BeaconConfig().MaxCommitteesPerSlot {
-			countAtSlot = params.BeaconConfig().MaxCommitteesPerSlot
-		}
-		if countAtSlot == 0 {
-			countAtSlot = 1
-		}
-		committeeItems := make([]*ethpb.BeaconCommittees_CommitteeItem, countAtSlot)
-		for i := uint64(0); i < countAtSlot; i++ {
-			epochOffset := i + (slot%params.BeaconConfig().SlotsPerEpoch)*countAtSlot
-			totalCount := countAtSlot * params.BeaconConfig().SlotsPerEpoch
-			committee, err := helpers.ComputeCommittee(activeIndices, attesterSeed, epochOffset, totalCount)
-			if err != nil {
-				t.Fatal(err)
-			}
-			committeeItems[i] = &ethpb.BeaconCommittees_CommitteeItem{
-				ValidatorIndices: committee,
-			}
-		}
-		wanted[slot] = &ethpb.BeaconCommittees_CommitteesList{
-			Committees: committeeItems,
-		}
+	wanted, err := computeCommittees(startSlot, activeIndices, attesterSeed)
+	if err != nil {
+		t.Fatal(err)
 	}
 
 	tests := []struct {
@@ -245,28 +195,9 @@ func TestServer_ListBeaconCommittees_FromArchive(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	wanted := make(map[uint64]*ethpb.BeaconCommittees_CommitteesList)
-	for slot := uint64(0); slot < params.BeaconConfig().SlotsPerEpoch; slot++ {
-		var countAtSlot = uint64(numValidators) / params.BeaconConfig().SlotsPerEpoch / params.BeaconConfig().TargetCommitteeSize
-		if countAtSlot > params.BeaconConfig().MaxCommitteesPerSlot {
-			countAtSlot = params.BeaconConfig().MaxCommitteesPerSlot
-		}
-		if countAtSlot == 0 {
-			countAtSlot = 1
-		}
-		committeeItems := make([]*ethpb.BeaconCommittees_CommitteeItem, countAtSlot)
-		for i := uint64(0); i < countAtSlot; i++ {
-			committee, err := helpers.BeaconCommittee(activeIndices, seed, slot, i)
-			if err != nil {
-				t.Fatal(err)
-			}
-			committeeItems[i] = &ethpb.BeaconCommittees_CommitteeItem{
-				ValidatorIndices: committee,
-			}
-		}
-		wanted[slot] = &ethpb.BeaconCommittees_CommitteesList{
-			Committees: committeeItems,
-		}
+	wanted, err := computeCommittees(0, activeIndices, seed)
+	if err != nil {
+		t.Fatal(err)
 	}
 	res1, err := bs.ListBeaconCommittees(context.Background(), &ethpb.ListCommitteesRequest{
 		QueryFilter: &ethpb.ListCommitteesRequest_Genesis{

shared/mock/beacon_chain_service_mock.go

@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: github.com/prysmaticlabs/ethereumapis/eth/v1alpha1 (interfaces: BeaconChainClient,BeaconChain_StreamChainHeadClient,BeaconChain_StreamBlocksClient,BeaconChain_StreamAttestationsClient,BeaconChain_StreamValidatorsInfoClient)
+// Source: github.com/prysmaticlabs/ethereumapis/eth/v1alpha1 (interfaces: BeaconChainClient,BeaconChain_StreamChainHeadClient,BeaconChain_StreamBlocksClient,BeaconChain_StreamAttestationsClient,BeaconChain_StreamValidatorsInfoClient,BeaconChain_StreamIndexedAttestationsClient)
 
 // Package mock is a generated GoMock package.
 package mock
@@ -257,6 +257,26 @@ func (mr *MockBeaconChainClientMockRecorder) ListBlocks(arg0, arg1 interface{},
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListBlocks", reflect.TypeOf((*MockBeaconChainClient)(nil).ListBlocks), varargs...)
 }
 
+// ListIndexedAttestations mocks base method
+func (m *MockBeaconChainClient) ListIndexedAttestations(arg0 context.Context, arg1 *v1alpha1.ListIndexedAttestationsRequest, arg2 ...grpc.CallOption) (*v1alpha1.ListIndexedAttestationsResponse, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{arg0, arg1}
+	for _, a := range arg2 {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "ListIndexedAttestations", varargs...)
+	ret0, _ := ret[0].(*v1alpha1.ListIndexedAttestationsResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ListIndexedAttestations indicates an expected call of ListIndexedAttestations
+func (mr *MockBeaconChainClientMockRecorder) ListIndexedAttestations(arg0, arg1 interface{}, arg2 ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{arg0, arg1}, arg2...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListIndexedAttestations", reflect.TypeOf((*MockBeaconChainClient)(nil).ListIndexedAttestations), varargs...)
+}
+
 // ListValidatorAssignments mocks base method
 func (m *MockBeaconChainClient) ListValidatorAssignments(arg0 context.Context, arg1 *v1alpha1.ListValidatorAssignmentsRequest, arg2 ...grpc.CallOption) (*v1alpha1.ValidatorAssignments, error) {
 	m.ctrl.T.Helper()
@@ -377,6 +397,26 @@ func (mr *MockBeaconChainClientMockRecorder) StreamChainHead(arg0, arg1 interfac
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StreamChainHead", reflect.TypeOf((*MockBeaconChainClient)(nil).StreamChainHead), varargs...)
 }
 
+// StreamIndexedAttestations mocks base method
+func (m *MockBeaconChainClient) StreamIndexedAttestations(arg0 context.Context, arg1 *empty.Empty, arg2 ...grpc.CallOption) (v1alpha1.BeaconChain_StreamIndexedAttestationsClient, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{arg0, arg1}
+	for _, a := range arg2 {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "StreamIndexedAttestations", varargs...)
+	ret0, _ := ret[0].(v1alpha1.BeaconChain_StreamIndexedAttestationsClient)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// StreamIndexedAttestations indicates an expected call of StreamIndexedAttestations
+func (mr *MockBeaconChainClientMockRecorder) StreamIndexedAttestations(arg0, arg1 interface{}, arg2 ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{arg0, arg1}, arg2...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StreamIndexedAttestations", reflect.TypeOf((*MockBeaconChainClient)(nil).StreamIndexedAttestations), varargs...)
+}
+
 // StreamValidatorsInfo mocks base method
 func (m *MockBeaconChainClient) StreamValidatorsInfo(arg0 context.Context, arg1 ...grpc.CallOption) (v1alpha1.BeaconChain_StreamValidatorsInfoClient, error) {
 	m.ctrl.T.Helper()
@@ -942,3 +982,126 @@ func (mr *MockBeaconChain_StreamValidatorsInfoClientMockRecorder) Trailer() *gom
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockBeaconChain_StreamValidatorsInfoClient)(nil).Trailer))
 }
+
+// MockBeaconChain_StreamIndexedAttestationsClient is a mock of BeaconChain_StreamIndexedAttestationsClient interface
+type MockBeaconChain_StreamIndexedAttestationsClient struct {
+	ctrl     *gomock.Controller
+	recorder *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder
+}
+
+// MockBeaconChain_StreamIndexedAttestationsClientMockRecorder is the mock recorder for MockBeaconChain_StreamIndexedAttestationsClient
+type MockBeaconChain_StreamIndexedAttestationsClientMockRecorder struct {
+	mock *MockBeaconChain_StreamIndexedAttestationsClient
+}
+
+// NewMockBeaconChain_StreamIndexedAttestationsClient creates a new mock instance
+func NewMockBeaconChain_StreamIndexedAttestationsClient(ctrl *gomock.Controller) *MockBeaconChain_StreamIndexedAttestationsClient {
+	mock := &MockBeaconChain_StreamIndexedAttestationsClient{ctrl: ctrl}
+	mock.recorder = &MockBeaconChain_StreamIndexedAttestationsClientMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) EXPECT() *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder {
+	return m.recorder
+}
+
+// CloseSend mocks base method
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) CloseSend() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CloseSend")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// CloseSend indicates an expected call of CloseSend
+func (mr *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder) CloseSend() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseSend", reflect.TypeOf((*MockBeaconChain_StreamIndexedAttestationsClient)(nil).CloseSend))
+}
+
+// Context mocks base method
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) Context() context.Context {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Context")
+	ret0, _ := ret[0].(context.Context)
+	return ret0
+}
+
+// Context indicates an expected call of Context
+func (mr *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder) Context() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockBeaconChain_StreamIndexedAttestationsClient)(nil).Context))
+}
+
+// Header mocks base method
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) Header() (metadata.MD, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Header")
+	ret0, _ := ret[0].(metadata.MD)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Header indicates an expected call of Header
+func (mr *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder) Header() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Header", reflect.TypeOf((*MockBeaconChain_StreamIndexedAttestationsClient)(nil).Header))
+}
+
+// Recv mocks base method
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) Recv() (*v1alpha1.IndexedAttestation, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Recv")
+	ret0, _ := ret[0].(*v1alpha1.IndexedAttestation)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Recv indicates an expected call of Recv
+func (mr *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder) Recv() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Recv", reflect.TypeOf((*MockBeaconChain_StreamIndexedAttestationsClient)(nil).Recv))
+}
+
+// RecvMsg mocks base method
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) RecvMsg(arg0 interface{}) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RecvMsg", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RecvMsg indicates an expected call of RecvMsg
+func (mr *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder) RecvMsg(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockBeaconChain_StreamIndexedAttestationsClient)(nil).RecvMsg), arg0)
+}
+
+// SendMsg mocks base method
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) SendMsg(arg0 interface{}) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SendMsg", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SendMsg indicates an expected call of SendMsg
+func (mr *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder) SendMsg(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockBeaconChain_StreamIndexedAttestationsClient)(nil).SendMsg), arg0)
+}
+
+// Trailer mocks base method
+func (m *MockBeaconChain_StreamIndexedAttestationsClient) Trailer() metadata.MD {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Trailer")
+	ret0, _ := ret[0].(metadata.MD)
+	return ret0
+}
+
+// Trailer indicates an expected call of Trailer
+func (mr *MockBeaconChain_StreamIndexedAttestationsClientMockRecorder) Trailer() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockBeaconChain_StreamIndexedAttestationsClient)(nil).Trailer))
+}

third_party/com_github_prysmaticlabs_ethereumapis-tags.patch

@@ -262,7 +262,7 @@ index 2ce5c34..4cbb276 100644
 +    bytes signature = 3 [(gogoproto.moretags) = "ssz-size:\"96\""];
  }
 diff --git a/eth/v1alpha1/beacon_chain.proto b/eth/v1alpha1/beacon_chain.proto
-index cdac301..945f8b5 100644
+index 0099328..8b8c6eb 100644
 --- a/eth/v1alpha1/beacon_chain.proto
 +++ b/eth/v1alpha1/beacon_chain.proto
 @@ -15,6 +15,7 @@ syntax = "proto3";
@@ -273,7 +273,7 @@ index cdac301..945f8b5 100644
  import "google/api/annotations.proto";
  import "google/protobuf/empty.proto";
  import "google/protobuf/any.proto";
-@@ -358,7 +359,7 @@ message ChainHead {
+@@ -410,7 +411,7 @@ message ChainHead {
      uint64 head_epoch = 2;
  
      // 32 byte merkle tree root of the canonical head block in the beacon node.
@@ -282,7 +282,7 @@ index cdac301..945f8b5 100644
  
      // Most recent slot that contains the finalized block.
      uint64 finalized_slot = 4;
-@@ -367,7 +368,7 @@ message ChainHead {
+@@ -419,7 +420,7 @@ message ChainHead {
      uint64 finalized_epoch = 5;
      
      // Most recent 32 byte finalized block root.
@@ -291,7 +291,7 @@ index cdac301..945f8b5 100644
  
      // Most recent slot that contains the justified block.
      uint64 justified_slot = 7;
-@@ -376,7 +377,7 @@ message ChainHead {
+@@ -428,7 +429,7 @@ message ChainHead {
      uint64 justified_epoch = 8;
      
      // Most recent 32 byte justified block root.
@@ -300,7 +300,7 @@ index cdac301..945f8b5 100644
  
      // Most recent slot that contains the previous justified block.
      uint64 previous_justified_slot = 10;
-@@ -385,7 +386,7 @@ message ChainHead {
+@@ -437,7 +438,7 @@ message ChainHead {
      uint64 previous_justified_epoch = 11;
  
      // Previous 32 byte justified block root.
@@ -309,7 +309,7 @@ index cdac301..945f8b5 100644
  }
  
  message ListCommitteesRequest {
-@@ -430,7 +431,7 @@ message ListValidatorBalancesRequest {
+@@ -482,7 +483,7 @@ message ListValidatorBalancesRequest {
  
      // Validator 48 byte BLS public keys to filter validators for the given
      // epoch.
@@ -318,7 +318,7 @@ index cdac301..945f8b5 100644
          
      // Validator indices to filter validators for the given epoch.
      repeated uint64 indices = 4;
-@@ -451,7 +452,7 @@ message ValidatorBalances {
+@@ -503,7 +504,7 @@ message ValidatorBalances {
  
      message Balance {
          // Validator's 48 byte BLS public key.
@@ -327,7 +327,7 @@ index cdac301..945f8b5 100644
  
          // Validator's index in the validator set.
          uint64 index = 2;
-@@ -500,7 +501,7 @@ message GetValidatorRequest {
+@@ -552,7 +553,7 @@ message GetValidatorRequest {
          uint64 index = 1;
  
          // 48 byte validator public key.
@@ -336,7 +336,7 @@ index cdac301..945f8b5 100644
      }
  }
  
-@@ -542,26 +543,25 @@ message ActiveSetChanges {
+@@ -594,26 +595,25 @@ message ActiveSetChanges {
      uint64 epoch = 1;
  
      // 48 byte validator public keys that have been activated in the given epoch.
@@ -369,7 +369,7 @@ index cdac301..945f8b5 100644
  
      // Indices of validators ejected in the given epoch.
      repeated uint64 ejected_indices = 9;
-@@ -611,11 +611,11 @@ message ValidatorQueue {
+@@ -663,11 +663,11 @@ message ValidatorQueue {
  
      // Ordered list of 48 byte public keys awaiting activation. 0th index is the
      // next key to be processed.
@@ -383,7 +383,7 @@ index cdac301..945f8b5 100644
  }
  
  message ListValidatorAssignmentsRequest {
-@@ -627,7 +627,7 @@ message ListValidatorAssignmentsRequest {
+@@ -679,7 +679,7 @@ message ListValidatorAssignmentsRequest {
          bool genesis = 2;
      }
      // 48 byte validator public keys to filter assignments for the given epoch.
@@ -392,7 +392,7 @@ index cdac301..945f8b5 100644
          
      // Validator indicies to filter assignments for the given epoch.
      repeated uint64 indices = 4;
-@@ -662,7 +662,7 @@ message ValidatorAssignments {
+@@ -714,7 +714,7 @@ message ValidatorAssignments {
          uint64 proposer_slot = 4;
  
          // 48 byte BLS public key.