Caplin: get aggregated attestation api (erigontech#9878)

https://ethereum.github.io/beacon-APIs/#/Validator/getAggregatedAttestation

cl/beacon/handler/handler.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/sentinel"
 	"github.com/ledgerwatch/erigon-lib/kv"
+	"github.com/ledgerwatch/erigon/cl/aggregation"
 	"github.com/ledgerwatch/erigon/cl/beacon/beacon_router_configuration"
 	"github.com/ledgerwatch/erigon/cl/beacon/beaconevents"
 	"github.com/ledgerwatch/erigon/cl/beacon/beaconhttp"
@@ -72,6 +73,7 @@ type ApiHandler struct {
 	syncMessagePool     sync_contribution_pool.SyncContributionPool
 	committeeSub        *committee_subscription.CommitteeSubscribeMgmt
 	attestationProducer attestation_producer.AttestationDataProducer
+	aggregatePool       aggregation.AggregationPool
 }
 
 func NewApiHandler(
@@ -95,6 +97,7 @@ func NewApiHandler(
 	engine execution_client.ExecutionEngine,
 	syncMessagePool sync_contribution_pool.SyncContributionPool,
 	committeeSub *committee_subscription.CommitteeSubscribeMgmt,
+	aggregatePool aggregation.AggregationPool,
 ) *ApiHandler {
 	blobBundles, err := lru.New[common.Bytes48, BlobBundle]("blobs", maxBlobBundleCacheSize)
 	if err != nil {
@@ -126,6 +129,7 @@ func NewApiHandler(
 		engine:              engine,
 		syncMessagePool:     syncMessagePool,
 		committeeSub:        committeeSub,
+		aggregatePool:       aggregatePool,
 	}
 }
 
@@ -239,7 +243,7 @@ func (a *ApiHandler) init() {
 					})
 					r.Get("/blinded_blocks/{slot}", http.NotFound)
 					r.Get("/attestation_data", beaconhttp.HandleEndpointFunc(a.GetEthV1ValidatorAttestationData))
-					r.Get("/aggregate_attestation", http.NotFound)
+					r.Get("/aggregate_attestation", beaconhttp.HandleEndpointFunc(a.GetEthV1ValidatorAggregateAttestation))
 					r.Post("/aggregate_and_proofs", a.PostEthV1ValidatorAggregatesAndProof)
 					r.Post("/beacon_committee_subscriptions", a.PostEthV1ValidatorBeaconCommitteeSubscription)
 					r.Post("/sync_committee_subscriptions", a.PostEthV1ValidatorSyncCommitteeSubscriptions)

cl/beacon/handler/utils_test.go

@@ -99,7 +99,7 @@ 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) // TODO: add tests
+		}, nil, blobStorage, nil, vp, nil, nil, fcu.SyncContributionPool, nil, nil) // TODO: add tests
 	h.Init()
 	return
 }

cl/beacon/handler/validator_test.go

@@ -0,0 +1,167 @@
+package handler
+
+import (
+	"encoding/json"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	libcommon "github.com/ledgerwatch/erigon-lib/common"
+	mock_aggregation "github.com/ledgerwatch/erigon/cl/aggregation/mock"
+	"github.com/ledgerwatch/erigon/cl/beacon/beacon_router_configuration"
+	"github.com/ledgerwatch/erigon/cl/cltypes/solid"
+	"github.com/ledgerwatch/erigon/cl/pool"
+	"github.com/ledgerwatch/erigon/common"
+	"github.com/stretchr/testify/suite"
+	"go.uber.org/mock/gomock"
+)
+
+type validatorTestSuite struct {
+	suite.Suite
+	apiHandler   *ApiHandler
+	mockAggrPool *mock_aggregation.MockAggregationPool
+	gomockCtrl   *gomock.Controller
+}
+
+func (t *validatorTestSuite) SetupTest() {
+	gomockCtrl := gomock.NewController(t.T())
+	t.mockAggrPool = mock_aggregation.NewMockAggregationPool(gomockCtrl)
+	t.apiHandler = NewApiHandler(
+		nil,
+		nil,
+		nil,
+		nil,
+		nil,
+		pool.OperationsPool{},
+		nil,
+		nil,
+		nil,
+		nil,
+		"0",
+		&beacon_router_configuration.RouterConfiguration{Validator: true},
+		nil,
+		nil,
+		nil,
+		nil,
+		nil,
+		nil,
+		nil,
+		nil,
+		t.mockAggrPool,
+	)
+	t.gomockCtrl = gomockCtrl
+}
+
+func (t *validatorTestSuite) TearDownTest() {
+	t.gomockCtrl.Finish()
+}
+
+func (t *validatorTestSuite) TestGetEthV1ValidatorAggregateAttestation() {
+	mockDataRoot := libcommon.HexToHash("0x123").String()
+
+	tests := []struct {
+		name    string
+		method  string
+		url     string
+		mock    func()
+		expCode int
+		expBody any
+	}{
+		{
+			name:    "empty attestation data root",
+			method:  http.MethodGet,
+			url:     "/eth/v1/validator/aggregate_attestation?slot=1",
+			mock:    func() {},
+			expCode: http.StatusBadRequest,
+			expBody: map[string]any{
+				"code":    float64(http.StatusBadRequest),
+				"message": "attestation_data_root is required",
+			},
+		},
+		{
+			name:   "slot mismatch",
+			method: http.MethodGet,
+			url:    "/eth/v1/validator/aggregate_attestation?attestation_data_root=" + mockDataRoot + "&slot=1",
+			mock: func() {
+				ret := *solid.NewAttestionFromParameters(
+					[]byte{},
+					solid.NewAttestionDataFromParameters(
+						123456,
+						1,
+						libcommon.HexToHash(mockDataRoot),
+						solid.NewCheckpointFromParameters(libcommon.Hash{}, 1),
+						solid.NewCheckpointFromParameters(libcommon.Hash{}, 1),
+					),
+					[96]byte{},
+				)
+				t.mockAggrPool.EXPECT().GetAggregatationByRoot(libcommon.HexToHash(mockDataRoot)).Return(&ret).Times(1)
+			},
+			expCode: http.StatusBadRequest,
+			expBody: map[string]any{
+				"code":    float64(http.StatusBadRequest),
+				"message": "attestation slot mismatch",
+			},
+		},
+		{
+			name:   "pass",
+			method: http.MethodGet,
+			url:    "/eth/v1/validator/aggregate_attestation?attestation_data_root=" + mockDataRoot + "&slot=1",
+			mock: func() {
+				ret := *solid.NewAttestionFromParameters(
+					[]byte{0b00111111, 0b00000011, 0, 0},
+					solid.NewAttestionDataFromParameters(
+						1,
+						1,
+						libcommon.HexToHash(mockDataRoot),
+						solid.NewCheckpointFromParameters(libcommon.Hash{}, 1),
+						solid.NewCheckpointFromParameters(libcommon.Hash{}, 1),
+					),
+					[96]byte{0, 1, 2, 3, 4, 5},
+				)
+				t.mockAggrPool.EXPECT().GetAggregatationByRoot(libcommon.HexToHash(mockDataRoot)).Return(&ret).Times(1)
+			},
+			expCode: http.StatusOK,
+			expBody: map[string]any{
+				"data": map[string]any{
+					"aggregation_bits": "0x" + common.Bytes2Hex([]byte{0b00111111, 0b00000011, 0, 0}),
+					"signature":        "0x" + common.Bytes2Hex([][96]byte{{0, 1, 2, 3, 4, 5}}[0][:]),
+					"data": map[string]any{
+						"slot":              "1",
+						"index":             "1",
+						"beacon_block_root": mockDataRoot,
+						"source": map[string]any{
+							"epoch": "1",
+							"root":  libcommon.Hash{}.String(),
+						},
+						"target": map[string]any{
+							"epoch": "1",
+							"root":  libcommon.Hash{}.String(),
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tc := range tests {
+		log.Printf("test case: %s", tc.name)
+		tc.mock()
+		req, err := http.NewRequest(tc.method, tc.url, nil)
+		t.NoError(err)
+		rr := httptest.NewRecorder()
+		t.apiHandler.ServeHTTP(rr, req)
+		t.Equal(tc.expCode, rr.Code)
+
+		// check body by comparing map
+		jsonResp := map[string]any{}
+		err = json.Unmarshal(rr.Body.Bytes(), &jsonResp)
+		t.NoError(err)
+		t.Equal(tc.expBody, jsonResp)
+
+		t.True(t.gomockCtrl.Satisfied(), "mock expectations were not met")
+	}
+}
+
+func TestValidator(t *testing.T) {
+	suite.Run(t, new(validatorTestSuite))
+}

cl/beacon/handler/validators.go

@@ -17,6 +17,7 @@ import (
 	state_accessors "github.com/ledgerwatch/erigon/cl/persistence/state"
 	"github.com/ledgerwatch/erigon/cl/phase1/core/state"
 	"github.com/ledgerwatch/log/v3"
+	"github.com/pkg/errors"
 	"golang.org/x/exp/slices"
 )
 
@@ -653,3 +654,30 @@ func shouldStatusBeFiltered(status validatorStatus, statuses []validatorStatus)
 	}
 	return true // filter if no filter condition is met
 }
+
+func (a *ApiHandler) GetEthV1ValidatorAggregateAttestation(w http.ResponseWriter, r *http.Request) (*beaconhttp.BeaconResponse, error) {
+	attDataRoot := r.URL.Query().Get("attestation_data_root")
+	if attDataRoot == "" {
+		return nil, beaconhttp.NewEndpointError(http.StatusBadRequest, fmt.Errorf("attestation_data_root is required"))
+	}
+	slot := r.URL.Query().Get("slot")
+	if slot == "" {
+		return nil, beaconhttp.NewEndpointError(http.StatusBadRequest, fmt.Errorf("slot is required"))
+	}
+	slotNum, err := strconv.ParseUint(slot, 10, 64)
+	if err != nil {
+		return nil, beaconhttp.NewEndpointError(http.StatusBadRequest, errors.WithMessage(err, "invalid slot"))
+	}
+
+	attDataRootHash := libcommon.HexToHash(attDataRoot)
+	att := a.aggregatePool.GetAggregatationByRoot(attDataRootHash)
+	if att == nil {
+		return nil, beaconhttp.NewEndpointError(http.StatusNotFound, fmt.Errorf("attestation not found. attestation_data_root"))
+	}
+	if slotNum != att.AttestantionData().Slot() {
+		log.Debug("attestation slot does not match", "attestation_data_root", attDataRoot, "slot_inquire", slot)
+		return nil, beaconhttp.NewEndpointError(http.StatusBadRequest, fmt.Errorf("attestation slot mismatch"))
+	}
+
+	return newBeaconResponse(att), nil
+}

cmd/caplin/caplin1/run.go

@@ -258,7 +258,7 @@ func RunCaplinPhase1(ctx context.Context, engine execution_client.ExecutionEngin
 	statesReader := historical_states_reader.NewHistoricalStatesReader(beaconConfig, rcsn, vTables, genesisState)
 	validatorParameters := validator_params.NewValidatorParams()
 	if config.BeaconRouter.Active {
-		apiHandler := handler.NewApiHandler(logger, genesisConfig, beaconConfig, indexDB, forkChoice, pool, rcsn, syncedDataManager, statesReader, sentinel, params.GitTag, &config.BeaconRouter, emitters, blobStorage, csn, validatorParameters, attestationProducer, engine, syncContributionPool, committeeSub)
+		apiHandler := handler.NewApiHandler(logger, genesisConfig, beaconConfig, indexDB, forkChoice, pool, rcsn, syncedDataManager, statesReader, sentinel, params.GitTag, &config.BeaconRouter, emitters, blobStorage, csn, validatorParameters, attestationProducer, engine, syncContributionPool, committeeSub, aggregationPool)
 		go beacon.ListenAndServe(&beacon.LayeredBeaconHandler{
 			ArchiveApi: apiHandler,
 		}, config.BeaconRouter)