move more services out from ForkchoiceStore (#9981)

- voluntary_exit
- bls_to_execution_change
- proposer_slashing
- expirable lru

---------

Co-authored-by: Giulio <giulio.rebuffo@gmail.com>

cl/beacon/handler/handler.go

@@ -83,6 +83,9 @@ type ApiHandler struct {
 	syncContributionAndProofsService services.SyncContributionService
 	aggregateAndProofsService        services.AggregateAndProofService
 	attestationService               services.AttestationService
+	voluntaryExitService             services.VoluntaryExitService
+	blsToExecutionChangeService      services.BLSToExecutionChangeService
+	proposerSlashingService          services.ProposerSlashingService
 }
 
 func NewApiHandler(
@@ -112,6 +115,9 @@ func NewApiHandler(
 	syncContributionAndProofs services.SyncContributionService,
 	aggregateAndProofs services.AggregateAndProofService,
 	attestationService services.AttestationService,
+	voluntaryExitService services.VoluntaryExitService,
+	blsToExecutionChangeService services.BLSToExecutionChangeService,
+	proposerSlashingService services.ProposerSlashingService,
 ) *ApiHandler {
 	blobBundles, err := lru.New[common.Bytes48, BlobBundle]("blobs", maxBlobBundleCacheSize)
 	if err != nil {
@@ -149,6 +155,9 @@ func NewApiHandler(
 		syncContributionAndProofsService: syncContributionAndProofs,
 		aggregateAndProofsService:        aggregateAndProofs,
 		attestationService:               attestationService,
+		voluntaryExitService:             voluntaryExitService,
+		blsToExecutionChangeService:      blsToExecutionChangeService,
+		proposerSlashingService:          proposerSlashingService,
 	}
 }
 

cl/beacon/handler/pool.go

@@ -17,7 +17,7 @@ import (
 )
 
 func (a *ApiHandler) GetEthV1BeaconPoolVoluntaryExits(w http.ResponseWriter, r *http.Request) (*beaconhttp.BeaconResponse, error) {
-	return newBeaconResponse(a.operationsPool.VoluntaryExistsPool.Raw()), nil
+	return newBeaconResponse(a.operationsPool.VoluntaryExitPool.Raw()), nil
 }
 
 func (a *ApiHandler) GetEthV1BeaconPoolAttesterSlashings(w http.ResponseWriter, r *http.Request) (*beaconhttp.BeaconResponse, error) {
@@ -125,10 +125,11 @@ func (a *ApiHandler) PostEthV1BeaconPoolVoluntaryExits(w http.ResponseWriter, r
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
-	if err := a.forkchoiceStore.OnVoluntaryExit(&req, false); err != nil {
+	if err := a.voluntaryExitService.ProcessMessage(r.Context(), nil, &req); err != nil && !errors.Is(err, services.ErrIgnore) {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
+
 	// Broadcast to gossip
 	if a.sentinel != nil {
 		encodedSSZ, err := req.EncodeSSZ(nil)
@@ -143,7 +144,7 @@ func (a *ApiHandler) PostEthV1BeaconPoolVoluntaryExits(w http.ResponseWriter, r
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
 		}
-		a.operationsPool.VoluntaryExistsPool.Insert(req.VoluntaryExit.ValidatorIndex, &req)
+		a.operationsPool.VoluntaryExitPool.Insert(req.VoluntaryExit.ValidatorIndex, &req)
 	}
 	// Only write 200
 	w.WriteHeader(http.StatusOK)
@@ -184,7 +185,7 @@ func (a *ApiHandler) PostEthV1BeaconPoolProposerSlashings(w http.ResponseWriter,
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
-	if err := a.forkchoiceStore.OnProposerSlashing(&req, false); err != nil {
+	if err := a.proposerSlashingService.ProcessMessage(r.Context(), nil, &req); err != nil && !errors.Is(err, services.ErrIgnore) {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
@@ -226,7 +227,7 @@ func (a *ApiHandler) PostEthV1BeaconPoolBlsToExecutionChanges(w http.ResponseWri
 	}
 	failures := []poolingFailure{}
 	for _, v := range req {
-		if err := a.forkchoiceStore.OnBlsToExecutionChange(v, false); err != nil {
+		if err := a.blsToExecutionChangeService.ProcessMessage(r.Context(), nil, v); err != nil && !errors.Is(err, services.ErrIgnore) {
 			failures = append(failures, poolingFailure{Index: len(failures), Message: err.Error()})
 			continue
 		}

cl/beacon/handler/utils_test.go

@@ -88,6 +88,10 @@ func setupTestingHandler(t *testing.T, v clparams.StateVersion, logger log.Logge
 	syncCommitteeMessagesService := mock_services.NewMockSyncCommitteeMessagesService(ctrl)
 	syncContributionService := mock_services.NewMockSyncContributionService(ctrl)
 	aggregateAndProofsService := mock_services.NewMockAggregateAndProofService(ctrl)
+	voluntaryExitService := mock_services.NewMockVoluntaryExitService(ctrl)
+	blsToExecutionChangeService := mock_services.NewMockBLSToExecutionChangeService(ctrl)
+	proposerSlashingService := mock_services.NewMockProposerSlashingService(ctrl)
+
 	// ctx context.Context, subnetID *uint64, msg *cltypes.SyncCommitteeMessage) error
 	syncCommitteeMessagesService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SyncCommitteeMessage) error {
 		return h.syncMessagePool.AddSyncCommitteeMessage(postState, *subnetID, msg)
@@ -96,11 +100,22 @@ func setupTestingHandler(t *testing.T, v clparams.StateVersion, logger log.Logge
 	syncContributionService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedContributionAndProof) error {
 		return h.syncMessagePool.AddSyncContribution(postState, msg.Message.Contribution)
 	}).AnyTimes()
-
 	aggregateAndProofsService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedAggregateAndProof) error {
 		opPool.AttestationsPool.Insert(msg.Message.Aggregate.Signature(), msg.Message.Aggregate)
 		return nil
 	}).AnyTimes()
+	voluntaryExitService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedVoluntaryExit) error {
+		opPool.VoluntaryExitPool.Insert(msg.VoluntaryExit.ValidatorIndex, msg)
+		return nil
+	}).AnyTimes()
+	blsToExecutionChangeService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedBLSToExecutionChange) error {
+		opPool.BLSToExecutionChangesPool.Insert(msg.Signature, msg)
+		return nil
+	}).AnyTimes()
+	proposerSlashingService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.ProposerSlashing) error {
+		opPool.ProposerSlashingsPool.Insert(pool.ComputeKeyForProposerSlashing(msg), msg)
+		return nil
+	}).AnyTimes()
 
 	vp = validator_params.NewValidatorParams()
 	h = NewApiHandler(
@@ -124,7 +139,15 @@ func setupTestingHandler(t *testing.T, v clparams.StateVersion, logger log.Logge
 			Events:     true,
 			Validator:  true,
 			Lighthouse: true,
-		}, nil, blobStorage, nil, vp, nil, nil, fcu.SyncContributionPool, nil, nil, syncCommitteeMessagesService, syncContributionService, aggregateAndProofsService, nil) // TODO: add tests
+		}, nil, blobStorage, nil, vp, nil, nil, fcu.SyncContributionPool, nil, nil,
+		syncCommitteeMessagesService,
+		syncContributionService,
+		aggregateAndProofsService,
+		nil,
+		voluntaryExitService,
+		blsToExecutionChangeService,
+		proposerSlashingService,
+	) // TODO: add tests
 	h.Init()
 	return
 }

cl/beacon/handler/validator_test.go

@@ -54,6 +54,9 @@ func (t *validatorTestSuite) SetupTest() {
 		nil,
 		nil,
 		nil,
+		nil,
+		nil,
+		nil,
 	)
 	t.gomockCtrl = gomockCtrl
 }

cl/phase1/core/state/lru/lru.go

@@ -2,8 +2,10 @@ package lru
 
 import (
 	"fmt"
+	"time"
 
 	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/hashicorp/golang-lru/v2/expirable"
 	"github.com/ledgerwatch/erigon-lib/metrics"
 )
 
@@ -39,3 +41,23 @@ func (c *Cache[K, V]) Get(k K) (V, bool) {
 	}
 	return v, ok
 }
+
+type CacheWithTTL[K comparable, V any] struct {
+	*expirable.LRU[K, V]
+	metric string
+}
+
+func NewWithTTL[K comparable, V any](metricName string, size int, ttl time.Duration) *CacheWithTTL[K, V] {
+	cache := expirable.NewLRU[K, V](size, nil, ttl)
+	return &CacheWithTTL[K, V]{LRU: cache, metric: metricName}
+}
+
+func (c *CacheWithTTL[K, V]) Get(k K) (V, bool) {
+	v, ok := c.LRU.Get(k)
+	if ok {
+		metrics.GetOrCreateCounter(fmt.Sprintf(`golang_ttl_lru_cache_hit{%s="%s"}`, "cache", c.metric)).Inc()
+	} else {
+		metrics.GetOrCreateCounter(fmt.Sprintf(`golang_ttl_lru_cache_miss{%s="%s"}`, "cache", c.metric)).Inc()
+	}
+	return v, ok
+}

cl/phase1/forkchoice/fork_choice_test.go

@@ -107,22 +107,7 @@ func TestForkChoiceBasic(t *testing.T) {
 	for sd.HeadState() == nil {
 		time.Sleep(time.Millisecond)
 	}
-	// Try processing a voluntary exit
-	err = store.OnVoluntaryExit(&cltypes.SignedVoluntaryExit{
-		VoluntaryExit: &cltypes.VoluntaryExit{
-			Epoch:          0,
-			ValidatorIndex: 0,
-		},
-	}, true)
 	require.NoError(t, err)
-	// Try processing a bls execution change exit
-	err = store.OnBlsToExecutionChange(&cltypes.SignedBLSToExecutionChange{
-		Message: &cltypes.BLSToExecutionChange{
-			ValidatorIndex: 0,
-		},
-	}, true)
-	require.NoError(t, err)
-	require.Equal(t, len(pool.VoluntaryExistsPool.Raw()), 1)
 }
 
 func TestForkChoiceChainBellatrix(t *testing.T) {

cl/phase1/forkchoice/forkchoice_mock.go

@@ -182,21 +182,6 @@ func (f *ForkChoiceStorageMock) Partecipation(epoch uint64) (*solid.BitList, boo
 	return f.ParticipationVal, f.ParticipationVal != nil
 }
 
-func (f *ForkChoiceStorageMock) OnVoluntaryExit(signedVoluntaryExit *cltypes.SignedVoluntaryExit, test bool) error {
-	f.Pool.VoluntaryExistsPool.Insert(signedVoluntaryExit.VoluntaryExit.ValidatorIndex, signedVoluntaryExit)
-	return nil
-}
-
-func (f *ForkChoiceStorageMock) OnProposerSlashing(proposerSlashing *cltypes.ProposerSlashing, test bool) error {
-	f.Pool.ProposerSlashingsPool.Insert(pool.ComputeKeyForProposerSlashing(proposerSlashing), proposerSlashing)
-	return nil
-}
-
-func (f *ForkChoiceStorageMock) OnBlsToExecutionChange(signedChange *cltypes.SignedBLSToExecutionChange, test bool) error {
-	f.Pool.BLSToExecutionChangesPool.Insert(signedChange.Signature, signedChange)
-	return nil
-}
-
 func (f *ForkChoiceStorageMock) ForkNodes() []ForkNode {
 	return f.WeightsMock
 }

cl/phase1/forkchoice/interface.go

@@ -60,9 +60,6 @@ type ForkChoiceStorageReader interface {
 type ForkChoiceStorageWriter interface {
 	OnAttestation(attestation *solid.Attestation, fromBlock, insert bool) error
 	OnAttesterSlashing(attesterSlashing *cltypes.AttesterSlashing, test bool) error
-	OnVoluntaryExit(signedVoluntaryExit *cltypes.SignedVoluntaryExit, test bool) error
-	OnProposerSlashing(proposerSlashing *cltypes.ProposerSlashing, test bool) error
-	OnBlsToExecutionChange(signedChange *cltypes.SignedBLSToExecutionChange, test bool) error
 	OnBlock(ctx context.Context, block *cltypes.SignedBeaconBlock, newPayload bool, fullValidation bool, checkDataAvaibility bool) error
 	AddPreverifiedBlobSidecar(blobSidecar *cltypes.BlobSidecar) error
 	OnTick(time uint64)