Better attestation packing for Electra

CHANGELOG.md

@@ -16,6 +16,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Added SubmitPoolAttesterSlashingV2 endpoint.
 - Added SubmitAggregateAndProofsRequestV2 endpoint.
 - Updated the `beacon-chain/monitor` package to Electra. [PR](https://github.com/prysmaticlabs/prysm/pull/14562)
+- Better attestation packing for Electra. [PR](https://github.com/prysmaticlabs/prysm/pull/14534)
 
 ### Changed
 

beacon-chain/operations/attestations/mock/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
     importpath = "github.com/prysmaticlabs/prysm/v5/beacon-chain/operations/attestations/mock",
     visibility = ["//visibility:public"],
     deps = [
+        "//beacon-chain/operations/attestations:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
     ],

beacon-chain/operations/attestations/mock/mock.go

@@ -3,13 +3,17 @@ package mock
 import (
 	"context"
 
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/operations/attestations"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 )
 
+var _ attestations.Pool = &PoolMock{}
+
 // PoolMock --
 type PoolMock struct {
-	AggregatedAtts []*ethpb.Attestation
+	AggregatedAtts   []ethpb.Att
+	UnaggregatedAtts []ethpb.Att
 }
 
 // AggregateUnaggregatedAttestations --
@@ -23,18 +27,18 @@ func (*PoolMock) AggregateUnaggregatedAttestationsBySlotIndex(_ context.Context,
 }
 
 // SaveAggregatedAttestation --
-func (*PoolMock) SaveAggregatedAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) SaveAggregatedAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 
 // SaveAggregatedAttestations --
-func (m *PoolMock) SaveAggregatedAttestations(atts []*ethpb.Attestation) error {
+func (m *PoolMock) SaveAggregatedAttestations(atts []ethpb.Att) error {
 	m.AggregatedAtts = append(m.AggregatedAtts, atts...)
 	return nil
 }
 
 // AggregatedAttestations --
-func (m *PoolMock) AggregatedAttestations() []*ethpb.Attestation {
+func (m *PoolMock) AggregatedAttestations() []ethpb.Att {
 	return m.AggregatedAtts
 }
 
@@ -43,13 +47,18 @@ func (*PoolMock) AggregatedAttestationsBySlotIndex(_ context.Context, _ primitiv
 	panic("implement me")
 }
 
+// AggregatedAttestationsBySlotIndexElectra --
+func (*PoolMock) AggregatedAttestationsBySlotIndexElectra(_ context.Context, _ primitives.Slot, _ primitives.CommitteeIndex) []*ethpb.AttestationElectra {
+	panic("implement me")
+}
+
 // DeleteAggregatedAttestation --
-func (*PoolMock) DeleteAggregatedAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) DeleteAggregatedAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 
 // HasAggregatedAttestation --
-func (*PoolMock) HasAggregatedAttestation(_ *ethpb.Attestation) (bool, error) {
+func (*PoolMock) HasAggregatedAttestation(_ ethpb.Att) (bool, error) {
 	panic("implement me")
 }
 
@@ -59,27 +68,33 @@ func (*PoolMock) AggregatedAttestationCount() int {
 }
 
 // SaveUnaggregatedAttestation --
-func (*PoolMock) SaveUnaggregatedAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) SaveUnaggregatedAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 
 // SaveUnaggregatedAttestations --
-func (*PoolMock) SaveUnaggregatedAttestations(_ []*ethpb.Attestation) error {
-	panic("implement me")
+func (m *PoolMock) SaveUnaggregatedAttestations(atts []ethpb.Att) error {
+	m.UnaggregatedAtts = append(m.UnaggregatedAtts, atts...)
+	return nil
 }
 
 // UnaggregatedAttestations --
-func (*PoolMock) UnaggregatedAttestations() ([]*ethpb.Attestation, error) {
-	panic("implement me")
+func (m *PoolMock) UnaggregatedAttestations() ([]ethpb.Att, error) {
+	return m.UnaggregatedAtts, nil
 }
 
 // UnaggregatedAttestationsBySlotIndex --
 func (*PoolMock) UnaggregatedAttestationsBySlotIndex(_ context.Context, _ primitives.Slot, _ primitives.CommitteeIndex) []*ethpb.Attestation {
 	panic("implement me")
 }
 
+// UnaggregatedAttestationsBySlotIndexElectra --
+func (*PoolMock) UnaggregatedAttestationsBySlotIndexElectra(_ context.Context, _ primitives.Slot, _ primitives.CommitteeIndex) []*ethpb.AttestationElectra {
+	panic("implement me")
+}
+
 // DeleteUnaggregatedAttestation --
-func (*PoolMock) DeleteUnaggregatedAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) DeleteUnaggregatedAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 
@@ -94,42 +109,42 @@ func (*PoolMock) UnaggregatedAttestationCount() int {
 }
 
 // SaveBlockAttestation --
-func (*PoolMock) SaveBlockAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) SaveBlockAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 
 // SaveBlockAttestations --
-func (*PoolMock) SaveBlockAttestations(_ []*ethpb.Attestation) error {
+func (*PoolMock) SaveBlockAttestations(_ []ethpb.Att) error {
 	panic("implement me")
 }
 
 // BlockAttestations --
-func (*PoolMock) BlockAttestations() []*ethpb.Attestation {
+func (*PoolMock) BlockAttestations() []ethpb.Att {
 	panic("implement me")
 }
 
 // DeleteBlockAttestation --
-func (*PoolMock) DeleteBlockAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) DeleteBlockAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 
 // SaveForkchoiceAttestation --
-func (*PoolMock) SaveForkchoiceAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) SaveForkchoiceAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 
 // SaveForkchoiceAttestations --
-func (*PoolMock) SaveForkchoiceAttestations(_ []*ethpb.Attestation) error {
+func (*PoolMock) SaveForkchoiceAttestations(_ []ethpb.Att) error {
 	panic("implement me")
 }
 
 // ForkchoiceAttestations --
-func (*PoolMock) ForkchoiceAttestations() []*ethpb.Attestation {
+func (*PoolMock) ForkchoiceAttestations() []ethpb.Att {
 	panic("implement me")
 }
 
 // DeleteForkchoiceAttestation --
-func (*PoolMock) DeleteForkchoiceAttestation(_ *ethpb.Attestation) error {
+func (*PoolMock) DeleteForkchoiceAttestation(_ ethpb.Att) error {
 	panic("implement me")
 }
 

beacon-chain/rpc/prysm/v1alpha1/validator/BUILD.bazel

@@ -212,7 +212,9 @@ go_test(
     embed = [":go_default_library"],
     eth_network = "minimal",
     tags = ["minimal"],
-    deps = common_deps,
+    deps = common_deps + [
+        "//beacon-chain/operations/attestations/mock:go_default_library",
+    ],
 )
 
 go_test(

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

@@ -91,14 +91,7 @@ func (vs *Server) packAttestations(ctx context.Context, latestState state.Beacon
 
 	var attsForInclusion proposerAtts
 	if postElectra {
-		// TODO: hack for Electra devnet-1, take only one aggregate per ID
-		// (which essentially means one aggregate for an attestation_data+committee combination
-		topAggregates := make([]ethpb.Att, 0)
-		for _, v := range attsById {
-			topAggregates = append(topAggregates, v[0])
-		}
-
-		attsForInclusion, err = computeOnChainAggregate(topAggregates)
+		attsForInclusion, err = onChainAggregates(attsById)
 		if err != nil {
 			return nil, err
 		}
@@ -113,14 +106,68 @@ func (vs *Server) packAttestations(ctx context.Context, latestState state.Beacon
 	if err != nil {
 		return nil, err
 	}
-	sorted, err := deduped.sort()
-	if err != nil {
-		return nil, err
+
+	var sorted proposerAtts
+	if postElectra {
+		sorted, err = deduped.sortOnChainAggregates()
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		sorted, err = deduped.sort()
+		if err != nil {
+			return nil, err
+		}
 	}
+
 	atts = sorted.limitToMaxAttestations()
 	return vs.filterAttestationBySignature(ctx, atts, latestState)
 }
 
+func onChainAggregates(attsById map[attestation.Id][]ethpb.Att) (proposerAtts, error) {
+	var result proposerAtts
+	var err error
+
+	// When constructing on-chain aggregates, we want to combine the most profitable
+	// aggregate for each ID, then the second most profitable, and so on and so forth.
+	// Because of this we sort attestations at the beginning.
+	for id, as := range attsById {
+		attsById[id], err = proposerAtts(as).sort()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// We construct the first on-chain aggregate by taking the first aggregate for each ID.
+	// We construct the second on-chain aggregate by taking the second aggregate for each ID.
+	// We continue doing this until we run out of aggregates.
+	idx := 0
+	for {
+		topAggregates := make([]ethpb.Att, 0, len(attsById))
+		for _, as := range attsById {
+			// In case there are no more aggregates for an ID, we skip that ID.
+			if len(as) > idx {
+				topAggregates = append(topAggregates, as[idx])
+			}
+		}
+
+		// Once there are no more aggregates for any ID, we are done.
+		if len(topAggregates) == 0 {
+			break
+		}
+
+		onChainAggs, err := computeOnChainAggregate(topAggregates)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, onChainAggs...)
+
+		idx++
+	}
+
+	return result, nil
+}
+
 // filter separates attestation list into two groups: valid and invalid attestations.
 // The first group passes the all the required checks for attestation to be considered for proposing.
 // And attestations from the second group should be deleted.
@@ -223,6 +270,14 @@ func (a proposerAtts) sort() (proposerAtts, error) {
 	return a.sortBySlotAndCommittee()
 }
 
+func (a proposerAtts) sortOnChainAggregates() (proposerAtts, error) {
+	if len(a) < 2 {
+		return a, nil
+	}
+
+	return a.sortByProfitabilityUsingMaxCover()
+}
+
 // Separate attestations by slot, as slot number takes higher precedence when sorting.
 // Also separate by committee index because maxcover will prefer attestations for the same
 // committee with disjoint bits over attestations for different committees with overlapping
@@ -231,7 +286,6 @@ func (a proposerAtts) sortBySlotAndCommittee() (proposerAtts, error) {
 	type slotAtts struct {
 		candidates map[primitives.CommitteeIndex]proposerAtts
 		selected   map[primitives.CommitteeIndex]proposerAtts
-		leftover   map[primitives.CommitteeIndex]proposerAtts
 	}
 
 	var slots []primitives.Slot
@@ -250,7 +304,6 @@ func (a proposerAtts) sortBySlotAndCommittee() (proposerAtts, error) {
 	var err error
 	for _, sa := range attsBySlot {
 		sa.selected = make(map[primitives.CommitteeIndex]proposerAtts)
-		sa.leftover = make(map[primitives.CommitteeIndex]proposerAtts)
 		for ci, committeeAtts := range sa.candidates {
 			sa.selected[ci], err = committeeAtts.sortByProfitabilityUsingMaxCover_committeeAwarePacking()
 			if err != nil {
@@ -266,9 +319,6 @@ func (a proposerAtts) sortBySlotAndCommittee() (proposerAtts, error) {
 	for _, slot := range slots {
 		sortedAtts = append(sortedAtts, sortSlotAttestations(attsBySlot[slot].selected)...)
 	}
-	for _, slot := range slots {
-		sortedAtts = append(sortedAtts, sortSlotAttestations(attsBySlot[slot].leftover)...)
-	}
 
 	return sortedAtts, nil
 }
@@ -287,15 +337,11 @@ func (a proposerAtts) sortByProfitabilityUsingMaxCover_committeeAwarePacking() (
 			return nil, err
 		}
 	}
-	// Add selected candidates on top, those that are not selected - append at bottom.
 	selectedKeys, _, err := aggregation.MaxCover(candidates, len(candidates), true /* allowOverlaps */)
 	if err != nil {
 		log.WithError(err).Debug("MaxCover aggregation failed")
 		return a, nil
 	}
-
-	// Pick selected attestations first, leftover attestations will be appended at the end.
-	// Both lists will be sorted by number of bits set.
 	selected := make(proposerAtts, selectedKeys.Count())
 	for i, key := range selectedKeys.BitIndices() {
 		selected[i] = a[key]

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

@@ -13,6 +13,9 @@ import (
 // computeOnChainAggregate constructs a final aggregate form a list of network aggregates with equal attestation data.
 // It assumes that each network aggregate has exactly one committee bit set.
 //
+// Our implementation allows to pass aggregates for different attestation data, in which case the function will return
+// one final aggregate per attestation data.
+//
 // Spec definition:
 //
 //	def compute_on_chain_aggregate(network_aggregates: Sequence[Attestation]) -> Attestation: