Caplin: Added SyncAggregate computation to block production (#10009)

Giulio2002 · web-flow · commit 6d9a5fd0e37b · 2024-04-23T15:31:52.000Z
This PR allows the computation for the computation of the `SyncAggregate` field in block production: https://sepolia.beaconcha.in/slot/4832922 proof of the code working is that now Caplin validators can include sync aggregates in their blocks. Things modified: * We do not aggregate pre-aggregated `SyncContributionAndProof`s, instead we just listen to the network and pick the most profitable ones for each sub sync committee (4 sync subcommittee on mainnet). profitability == most bits set in `AggregationBits` field. * Separate aggregates set for contribution to be included in a block from the ones constructed from `SyncCommitteeMessage`s, combining the two causes some contributions to be marked as invalid and not aggregable. * Remove SyncContributionMock in favor of gomock
diff --git a/cl/aggregation/pool_impl.go b/cl/aggregation/pool_impl.go
@@ -61,7 +61,7 @@ func (p *aggregationPoolImpl) AddAttestation(inAtt *solid.Attestation) error {
 		return nil
 	}
 
-	if utils.IsSupersetBitlist(att.AggregationBits(), inAtt.AggregationBits()) {
+	if utils.IsNonStrictSupersetBitlist(att.AggregationBits(), inAtt.AggregationBits()) {
 		return ErrIsSuperset
 	}
 
diff --git a/cl/beacon/handler/block_production.go b/cl/beacon/handler/block_production.go
@@ -11,7 +11,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/Giulio2002/bls"
 	"github.com/go-chi/chi/v5"
 	"github.com/ledgerwatch/erigon-lib/common"
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
@@ -170,10 +169,6 @@ func (a *ApiHandler) produceBeaconBody(ctx context.Context, apiVersion int, base
 	beaconBody.RandaoReveal = randaoReveal
 	beaconBody.Graffiti = graffiti
 	beaconBody.Version = stateVersion
-	// Sync aggregate is empty for now.
-	beaconBody.SyncAggregate = &cltypes.SyncAggregate{
-		SyncCommiteeSignature: bls.InfiniteSignature,
-	}
 
 	// Build execution payload
 	latestExecutionPayload := baseState.LatestExecutionPayloadHeader()
@@ -308,6 +303,15 @@ func (a *ApiHandler) produceBeaconBody(ctx context.Context, apiVersion int, base
 			}
 		}
 	}()
+	// process the sync aggregate in parallel
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		beaconBody.SyncAggregate, err = a.syncMessagePool.GetSyncAggregate(targetSlot-1, blockRoot)
+		if err != nil {
+			log.Error("BlockProduction: Failed to get sync aggregate", "err", err)
+		}
+	}()
 	wg.Wait()
 	if executionPayload == nil {
 		return nil, 0, fmt.Errorf("failed to produce execution payload")
diff --git a/cl/beacon/handler/utils_test.go b/cl/beacon/handler/utils_test.go
@@ -45,7 +45,7 @@ func setupTestingHandler(t *testing.T, v clparams.StateVersion, logger log.Logge
 		bcfg.CapellaForkEpoch = 1
 		blocks, preState, postState = tests.GetCapellaRandom()
 	}
-	fcu = forkchoice.NewForkChoiceStorageMock()
+	fcu = forkchoice.NewForkChoiceStorageMock(t)
 	db = memdb.NewTestDB(t)
 	blobDb := memdb.NewTestDB(t)
 	var reader *tests.MockBlockReader
diff --git a/cl/phase1/forkchoice/forkchoice_mock.go b/cl/phase1/forkchoice/forkchoice_mock.go
@@ -2,6 +2,7 @@ package forkchoice
 
 import (
 	"context"
+	"testing"
 
 	"github.com/ledgerwatch/erigon-lib/common"
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
@@ -12,6 +13,7 @@ import (
 	"github.com/ledgerwatch/erigon/cl/pool"
 	"github.com/ledgerwatch/erigon/cl/transition/impl/eth2"
 	"github.com/ledgerwatch/erigon/cl/validator/sync_contribution_pool"
+	"go.uber.org/mock/gomock"
 )
 
 // Make mocks with maps and simple setters and getters, panic on methods from ForkChoiceStorageWriter
@@ -47,7 +49,51 @@ type ForkChoiceStorageMock struct {
 	Pool pool.OperationsPool
 }
 
-func NewForkChoiceStorageMock() *ForkChoiceStorageMock {
+func makeSyncContributionPoolMock(t *testing.T) sync_contribution_pool.SyncContributionPool {
+	ctrl := gomock.NewController(t)
+	type syncContributionKey struct {
+		slot              uint64
+		subcommitteeIndex uint64
+		beaconBlockRoot   common.Hash
+	}
+	u := map[syncContributionKey]*cltypes.Contribution{}
+	pool := sync_contribution_pool.NewMockSyncContributionPool(ctrl)
+	pool.EXPECT().AddSyncContribution(gomock.Any(), gomock.Any()).DoAndReturn(func(headState *state.CachingBeaconState, contribution *cltypes.Contribution) error {
+		key := syncContributionKey{
+			slot:              contribution.Slot,
+			subcommitteeIndex: contribution.SubcommitteeIndex,
+			beaconBlockRoot:   contribution.BeaconBlockRoot,
+		}
+		u[key] = contribution
+		return nil
+	}).AnyTimes()
+	pool.EXPECT().GetSyncContribution(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(slot uint64, subcommitteeIndex uint64, beaconBlockRoot common.Hash) (*cltypes.Contribution, bool) {
+		key := syncContributionKey{
+			slot:              slot,
+			subcommitteeIndex: subcommitteeIndex,
+			beaconBlockRoot:   beaconBlockRoot,
+		}
+		v, ok := u[key]
+		return v, ok
+	}).AnyTimes()
+	pool.EXPECT().AddSyncCommitteeMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(headState *state.CachingBeaconState, subCommitee uint64, message *cltypes.SyncCommitteeMessage) error {
+		key := syncContributionKey{
+			slot:              message.Slot,
+			subcommitteeIndex: subCommitee,
+			beaconBlockRoot:   message.BeaconBlockRoot,
+		}
+		u[key] = &cltypes.Contribution{
+			Slot:              message.Slot,
+			SubcommitteeIndex: subCommitee,
+			BeaconBlockRoot:   message.BeaconBlockRoot,
+			AggregationBits:   make([]byte, cltypes.SyncCommitteeAggregationBitsSize),
+		}
+		return nil
+	}).AnyTimes()
+	return pool
+}
+
+func NewForkChoiceStorageMock(t *testing.T) *ForkChoiceStorageMock {
 	return &ForkChoiceStorageMock{
 		Ancestors:                 make(map[uint64]common.Hash),
 		AnchorSlotVal:             0,
@@ -66,9 +112,9 @@ func NewForkChoiceStorageMock() *ForkChoiceStorageMock {
 		GetFinalityCheckpointsVal: make(map[common.Hash][3]solid.Checkpoint),
 		LightClientBootstraps:     make(map[common.Hash]*cltypes.LightClientBootstrap),
 		LCUpdates:                 make(map[uint64]*cltypes.LightClientUpdate),
-		SyncContributionPool:      sync_contribution_pool.NewSyncContributionPoolMock(),
 		Headers:                   make(map[common.Hash]*cltypes.BeaconBlockHeader),
 		GetBeaconCommitteeMock:    nil,
+		SyncContributionPool:      makeSyncContributionPoolMock(t),
 	}
 }
 
diff --git a/cl/phase1/network/services/aggregate_and_proof_service_test.go b/cl/phase1/network/services/aggregate_and_proof_service_test.go
@@ -43,7 +43,7 @@ func setupAggregateAndProofTest(t *testing.T) (AggregateAndProofService, *synced
 	cn()
 	cfg := &clparams.MainnetBeaconConfig
 	syncedDataManager := synced_data.NewSyncedDataManager(true, cfg)
-	forkchoiceMock := forkchoice.NewForkChoiceStorageMock()
+	forkchoiceMock := forkchoice.NewForkChoiceStorageMock(t)
 	blockService := NewAggregateAndProofService(ctx, syncedDataManager, forkchoiceMock, cfg, nil, true)
 	return blockService, syncedDataManager, forkchoiceMock
 }
diff --git a/cl/phase1/network/services/blob_sidecar_service_test.go b/cl/phase1/network/services/blob_sidecar_service_test.go
@@ -56,7 +56,7 @@ func setupBlobSidecarService(t *testing.T, ctrl *gomock.Controller, test bool) (
 	cfg := &clparams.MainnetBeaconConfig
 	syncedDataManager := synced_data.NewSyncedDataManager(true, cfg)
 	ethClock := eth_clock.NewMockEthereumClock(ctrl)
-	forkchoiceMock := forkchoice.NewForkChoiceStorageMock()
+	forkchoiceMock := forkchoice.NewForkChoiceStorageMock(t)
 	blockService := NewBlobSidecarService(ctx2, cfg, forkchoiceMock, syncedDataManager, ethClock, test)
 	return blockService, syncedDataManager, ethClock, forkchoiceMock
 }
diff --git a/cl/phase1/network/services/block_service_test.go b/cl/phase1/network/services/block_service_test.go
@@ -21,7 +21,7 @@ func setupBlockService(t *testing.T, ctrl *gomock.Controller) (BlockService, *sy
 	cfg := &clparams.MainnetBeaconConfig
 	syncedDataManager := synced_data.NewSyncedDataManager(true, cfg)
 	ethClock := eth_clock.NewMockEthereumClock(ctrl)
-	forkchoiceMock := forkchoice.NewForkChoiceStorageMock()
+	forkchoiceMock := forkchoice.NewForkChoiceStorageMock(t)
 	blockService := NewBlockService(context.Background(), db, forkchoiceMock, syncedDataManager, ethClock, cfg, nil)
 	return blockService, syncedDataManager, ethClock, forkchoiceMock
 }
diff --git a/cl/phase1/network/services/sync_committee_messages_service.go b/cl/phase1/network/services/sync_committee_messages_service.go
@@ -99,7 +99,7 @@ func (s *syncCommitteeMessagesService) ProcessMessage(ctx context.Context, subne
 func (s *syncCommitteeMessagesService) cleanupOldSyncCommitteeMessages() {
 	headSlot := s.syncedDataManager.HeadSlot()
 	for k := range s.seenSyncCommitteeMessages {
-		if headSlot != k.slot {
+		if headSlot > k.slot+1 {
 			delete(s.seenSyncCommitteeMessages, k)
 		}
 	}
diff --git a/cl/sentinel/handlers/heartbeats_test.go b/cl/sentinel/handlers/heartbeats_test.go
@@ -68,7 +68,7 @@ func TestPing(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 	ethClock := getEthClock(t)
 
 	_, beaconCfg := clparams.GetConfigsByNetwork(1)
@@ -123,7 +123,7 @@ func TestGoodbye(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 	ethClock := getEthClock(t)
 	_, beaconCfg := clparams.GetConfigsByNetwork(1)
 	c := NewConsensusHandlers(
@@ -183,7 +183,7 @@ func TestMetadataV2(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 	ethClock := getEthClock(t)
 	nc := clparams.NetworkConfigs[clparams.MainnetNetwork]
 	_, beaconCfg := clparams.GetConfigsByNetwork(1)
@@ -241,7 +241,7 @@ func TestMetadataV1(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 
 	nc := clparams.NetworkConfigs[clparams.MainnetNetwork]
 	ethClock := getEthClock(t)
@@ -299,7 +299,7 @@ func TestStatus(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 
 	hs := handshake.New(ctx, getEthClock(t), &clparams.MainnetBeaconConfig, nil)
 	s := &cltypes.Status{
diff --git a/cl/sentinel/handlers/light_client_test.go b/cl/sentinel/handlers/light_client_test.go
@@ -45,7 +45,7 @@ func TestLightClientOptimistic(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 
 	f.NewestLCUpdate = &cltypes.LightClientUpdate{
 		AttestedHeader:    cltypes.NewLightClientHeader(clparams.AltairVersion),
@@ -115,7 +115,7 @@ func TestLightClientFinality(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 
 	f.NewestLCUpdate = &cltypes.LightClientUpdate{
 		AttestedHeader:    cltypes.NewLightClientHeader(clparams.AltairVersion),
@@ -188,7 +188,7 @@ func TestLightClientBootstrap(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 
 	f.NewestLCUpdate = &cltypes.LightClientUpdate{
 		AttestedHeader:    cltypes.NewLightClientHeader(clparams.AltairVersion),
@@ -270,7 +270,7 @@ func TestLightClientUpdates(t *testing.T) {
 	peersPool := peers.NewPool()
 	beaconDB, indiciesDB := setupStore(t)
 
-	f := forkchoice.NewForkChoiceStorageMock()
+	f := forkchoice.NewForkChoiceStorageMock(t)
 	ethClock := getEthClock(t)
 
 	up := &cltypes.LightClientUpdate{
diff --git a/cl/utils/bytes.go b/cl/utils/bytes.go
@@ -121,18 +121,32 @@ func IsBitOn(b []byte, idx int) bool {
 	return b[idx/8]&i == i
 }
 
-func IsSupersetBitlist(a, b []byte) bool {
+// IsNonStrictSupersetBitlist checks if bitlist 'a' is a non-strict superset of bitlist 'b'
+func IsNonStrictSupersetBitlist(a, b []byte) bool {
+	// Ensure 'a' is at least as long as 'b'
 	if len(a) < len(b) {
 		return false
 	}
-	for i := range b {
-		if a[i]&b[i] != b[i] {
+
+	// Check each bit in 'b' to ensure it is also set in 'a'
+	for i := 0; i < len(b); i++ {
+		if (a[i] & b[i]) != b[i] {
 			return false
 		}
 	}
+
+	// If all bits required by 'b' are present in 'a', return true
 	return true
 }
 
+func BitsOnCount(b []byte) int {
+	count := 0
+	for _, v := range b {
+		count += bits.OnesCount8(v)
+	}
+	return count
+}
+
 func MergeBitlists(a, b []byte) {
 	for i := range b {
 		a[i] |= b[i]
diff --git a/cl/validator/sync_contribution_pool/interface.go b/cl/validator/sync_contribution_pool/interface.go
@@ -18,4 +18,6 @@ type SyncContributionPool interface {
 
 	// GetSyncContribution retrieves a sync contribution from the pool.
 	GetSyncContribution(slot, subcommitteeIndex uint64, beaconBlockRoot common.Hash) *cltypes.Contribution
+	// Obtain the sync aggregate for the sync messages pointing to a given beacon block root.
+	GetSyncAggregate(slot uint64, beaconBlockRoot common.Hash) (*cltypes.SyncAggregate, error)
 }
diff --git a/cl/validator/sync_contribution_pool/mock.go b/cl/validator/sync_contribution_pool/mock.go
diff --git a/cl/validator/sync_contribution_pool/sync_contribution_pool.go b/cl/validator/sync_contribution_pool/sync_contribution_pool.go
diff --git a/cl/validator/sync_contribution_pool/sync_contribution_pool_mock.go b/cl/validator/sync_contribution_pool/sync_contribution_pool_mock.go
diff --git a/cl/validator/sync_contribution_pool/sync_contribution_pool_test.go b/cl/validator/sync_contribution_pool/sync_contribution_pool_test.go
diff --git a/cmd/caplin/caplin1/run.go b/cmd/caplin/caplin1/run.go
diff --git a/cmd/sentinel/main.go b/cmd/sentinel/main.go

Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,7 @@ func (p aggregationPoolImpl) AddAttestation(inAtt solid.Attestation) error {`
`61`	`61`	`return nil`
`62`	`62`	`}`
`63`	`63`
`64`		`- if utils.IsSupersetBitlist(att.AggregationBits(), inAtt.AggregationBits()) {`
	`64`	`+ if utils.IsNonStrictSupersetBitlist(att.AggregationBits(), inAtt.AggregationBits()) {`
`65`	`65`	`return ErrIsSuperset`
`66`	`66`	`}`
`67`	`67`
Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,7 @@ func setupTestingHandler(t *testing.T, v clparams.StateVersion, logger log.Logge`
`45`	`45`	`bcfg.CapellaForkEpoch = 1`
`46`	`46`	`blocks, preState, postState = tests.GetCapellaRandom()`
`47`	`47`	`}`
`48`		`- fcu = forkchoice.NewForkChoiceStorageMock()`
	`48`	`+ fcu = forkchoice.NewForkChoiceStorageMock(t)`
`49`	`49`	`db = memdb.NewTestDB(t)`
`50`	`50`	`blobDb := memdb.NewTestDB(t)`
`51`	`51`	`var reader *tests.MockBlockReader`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ func setupAggregateAndProofTest(t testing.T) (AggregateAndProofService, synced`
`43`	`43`	`cn()`
`44`	`44`	`cfg := &clparams.MainnetBeaconConfig`
`45`	`45`	`syncedDataManager := synced_data.NewSyncedDataManager(true, cfg)`
`46`		`- forkchoiceMock := forkchoice.NewForkChoiceStorageMock()`
	`46`	`+ forkchoiceMock := forkchoice.NewForkChoiceStorageMock(t)`
`47`	`47`	`blockService := NewAggregateAndProofService(ctx, syncedDataManager, forkchoiceMock, cfg, nil, true)`
`48`	`48`	`return blockService, syncedDataManager, forkchoiceMock`
`49`	`49`	`}`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,7 @@ func setupBlobSidecarService(t testing.T, ctrl gomock.Controller, test bool) (`
`56`	`56`	`cfg := &clparams.MainnetBeaconConfig`
`57`	`57`	`syncedDataManager := synced_data.NewSyncedDataManager(true, cfg)`
`58`	`58`	`ethClock := eth_clock.NewMockEthereumClock(ctrl)`
`59`		`- forkchoiceMock := forkchoice.NewForkChoiceStorageMock()`
	`59`	`+ forkchoiceMock := forkchoice.NewForkChoiceStorageMock(t)`
`60`	`60`	`blockService := NewBlobSidecarService(ctx2, cfg, forkchoiceMock, syncedDataManager, ethClock, test)`
`61`	`61`	`return blockService, syncedDataManager, ethClock, forkchoiceMock`
`62`	`62`	`}`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ func setupBlockService(t testing.T, ctrl gomock.Controller) (BlockService, *sy`
`21`	`21`	`cfg := &clparams.MainnetBeaconConfig`
`22`	`22`	`syncedDataManager := synced_data.NewSyncedDataManager(true, cfg)`
`23`	`23`	`ethClock := eth_clock.NewMockEthereumClock(ctrl)`
`24`		`- forkchoiceMock := forkchoice.NewForkChoiceStorageMock()`
	`24`	`+ forkchoiceMock := forkchoice.NewForkChoiceStorageMock(t)`
`25`	`25`	`blockService := NewBlockService(context.Background(), db, forkchoiceMock, syncedDataManager, ethClock, cfg, nil)`
`26`	`26`	`return blockService, syncedDataManager, ethClock, forkchoiceMock`
`27`	`27`	`}`
Original file line number	Diff line number	Diff line change
`@@ -99,7 +99,7 @@ func (s *syncCommitteeMessagesService) ProcessMessage(ctx context.Context, subne`
`99`	`99`	`func (s *syncCommitteeMessagesService) cleanupOldSyncCommitteeMessages() {`
`100`	`100`	`headSlot := s.syncedDataManager.HeadSlot()`
`101`	`101`	`for k := range s.seenSyncCommitteeMessages {`
`102`		`- if headSlot != k.slot {`
	`102`	`+ if headSlot > k.slot+1 {`
`103`	`103`	`delete(s.seenSyncCommitteeMessages, k)`
`104`	`104`	`}`
`105`	`105`	`}`