Caplin: Fixed Invalid reading for historical states

cl/antiquary/antiquary.go

@@ -249,7 +249,7 @@ func (a *Antiquary) antiquate(version uint8, from, to uint64) error {
 	}
 	// Notify bittorent to seed the new snapshots
 	if _, err := a.downloader.Add(a.ctx, &proto_downloader.AddRequest{Items: downloadItems}); err != nil {
-		return err
+		log.Warn("[Antiquary]: Failed to add items to bittorent", "err", err)
 	}
 
 	tx, err := a.mainDB.BeginRw(a.ctx)

cl/beacon/handler/handler.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/go-chi/chi/v5"
+	"github.com/ledgerwatch/erigon-lib/gointerfaces/sentinel"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon/cl/beacon/beaconhttp"
 	"github.com/ledgerwatch/erigon/cl/beacon/synced_data"
@@ -29,15 +30,16 @@ type ApiHandler struct {
 	operationsPool  pool.OperationsPool
 	syncedData      *synced_data.SyncedDataManager
 	stateReader     *historical_states_reader.HistoricalStatesReader
+	sentinel        sentinel.SentinelClient
 
 	// pools
 	randaoMixesPool sync.Pool
 }
 
-func NewApiHandler(genesisConfig *clparams.GenesisConfig, beaconChainConfig *clparams.BeaconChainConfig, source persistence.RawBeaconBlockChain, indiciesDB kv.RoDB, forkchoiceStore forkchoice.ForkChoiceStorage, operationsPool pool.OperationsPool, rcsn freezeblocks.BeaconSnapshotReader, syncedData *synced_data.SyncedDataManager, stateReader *historical_states_reader.HistoricalStatesReader) *ApiHandler {
+func NewApiHandler(genesisConfig *clparams.GenesisConfig, beaconChainConfig *clparams.BeaconChainConfig, source persistence.RawBeaconBlockChain, indiciesDB kv.RoDB, forkchoiceStore forkchoice.ForkChoiceStorage, operationsPool pool.OperationsPool, rcsn freezeblocks.BeaconSnapshotReader, syncedData *synced_data.SyncedDataManager, stateReader *historical_states_reader.HistoricalStatesReader, sentinel sentinel.SentinelClient) *ApiHandler {
 	return &ApiHandler{o: sync.Once{}, genesisCfg: genesisConfig, beaconChainCfg: beaconChainConfig, indiciesDB: indiciesDB, forkchoiceStore: forkchoiceStore, operationsPool: operationsPool, blockReader: rcsn, syncedData: syncedData, stateReader: stateReader, randaoMixesPool: sync.Pool{New: func() interface{} {
 		return solid.NewHashVector(int(beaconChainConfig.EpochsPerHistoricalVector))
-	}}}
+	}}, sentinel: sentinel}
 }
 
 func (a *ApiHandler) init() {
@@ -73,13 +75,17 @@ func (a *ApiHandler) init() {
 				r.Get("/genesis", beaconhttp.HandleEndpointFunc(a.getGenesis))
 				r.Get("/blinded_blocks/{block_id}", beaconhttp.HandleEndpointFunc(a.getBlindedBlock))
 				r.Route("/pool", func(r chi.Router) {
-					r.Post("/attestations", http.NotFound)
-					r.Get("/voluntary_exits", beaconhttp.HandleEndpointFunc(a.poolVoluntaryExits))
-					r.Get("/attester_slashings", beaconhttp.HandleEndpointFunc(a.poolAttesterSlashings))
-					r.Get("/proposer_slashings", beaconhttp.HandleEndpointFunc(a.poolProposerSlashings))
-					r.Get("/bls_to_execution_changes", beaconhttp.HandleEndpointFunc(a.poolBlsToExecutionChanges))
-					r.Get("/attestations", beaconhttp.HandleEndpointFunc(a.poolAttestations))
-					r.Post("/sync_committees", http.NotFound)
+					r.Get("/voluntary_exits", beaconhttp.HandleEndpointFunc(a.GetEthV1BeaconPoolVoluntaryExits))
+					r.Post("/voluntary_exits", a.PostEthV1BeaconPoolVoluntaryExits)
+					r.Get("/attester_slashings", beaconhttp.HandleEndpointFunc(a.GetEthV1BeaconPoolAttesterSlashings))
+					r.Post("/attester_slashings", a.PostEthV1BeaconPoolAttesterSlashings)
+					r.Get("/proposer_slashings", beaconhttp.HandleEndpointFunc(a.GetEthV1BeaconPoolProposerSlashings))
+					r.Post("/proposer_slashings", a.PostEthV1BeaconPoolProposerSlashings)
+					r.Get("/bls_to_execution_changes", beaconhttp.HandleEndpointFunc(a.GetEthV1BeaconPoolBLSExecutionChanges))
+					r.Post("/bls_to_execution_changes", a.PostEthV1BeaconPoolBlsToExecutionChanges)
+					r.Get("/attestations", beaconhttp.HandleEndpointFunc(a.GetEthV1BeaconPoolAttestations))
+					r.Post("/attestations", http.NotFound)    // TODO
+					r.Post("/sync_committees", http.NotFound) // TODO
 				})
 				r.Get("/node/syncing", http.NotFound)
 				r.Route("/states", func(r chi.Router) {

cl/beacon/handler/pool.go

@@ -1,25 +1,193 @@
 package handler
 
 import (
+	"encoding/json"
+	"fmt"
 	"net/http"
+
+	"github.com/ledgerwatch/erigon-lib/gointerfaces/sentinel"
+	"github.com/ledgerwatch/erigon/cl/beacon/beaconhttp"
+	"github.com/ledgerwatch/erigon/cl/cltypes"
+	"github.com/ledgerwatch/erigon/cl/gossip"
 )
 
-func (a *ApiHandler) poolVoluntaryExits(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
+func (a *ApiHandler) GetEthV1BeaconPoolVoluntaryExits(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
 	return newBeaconResponse(a.operationsPool.VoluntaryExistsPool.Raw()), nil
 }
 
-func (a *ApiHandler) poolAttesterSlashings(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
+func (a *ApiHandler) GetEthV1BeaconPoolAttesterSlashings(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
+	fmt.Println("GetEthV1BeaconPoolAttesterSlashings", a.operationsPool.AttesterSlashingsPool.Raw())
 	return newBeaconResponse(a.operationsPool.AttesterSlashingsPool.Raw()), nil
 }
 
-func (a *ApiHandler) poolProposerSlashings(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
+func (a *ApiHandler) GetEthV1BeaconPoolProposerSlashings(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
 	return newBeaconResponse(a.operationsPool.ProposerSlashingsPool.Raw()), nil
 }
 
-func (a *ApiHandler) poolBlsToExecutionChanges(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
+func (a *ApiHandler) GetEthV1BeaconPoolBLSExecutionChanges(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
 	return newBeaconResponse(a.operationsPool.BLSToExecutionChangesPool.Raw()), nil
 }
 
-func (a *ApiHandler) poolAttestations(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
-	return newBeaconResponse(a.operationsPool.AttestationsPool.Raw()), nil
+func (a *ApiHandler) GetEthV1BeaconPoolAttestations(w http.ResponseWriter, r *http.Request) (*beaconResponse, error) {
+	slot, err := uint64FromQueryParams(r, "slot")
+	if err != nil {
+		return nil, beaconhttp.NewEndpointError(http.StatusBadRequest, err.Error())
+	}
+	committeeIndex, err := uint64FromQueryParams(r, "committee_index")
+	if err != nil {
+		return nil, beaconhttp.NewEndpointError(http.StatusBadRequest, err.Error())
+	}
+	atts := a.operationsPool.AttestationsPool.Raw()
+	if slot == nil && committeeIndex == nil {
+		return newBeaconResponse(atts), nil
+	}
+	ret := make([]any, 0, len(atts))
+	for i := range atts {
+		if slot != nil && atts[i].AttestantionData().Slot() != *slot {
+			continue
+		}
+		if committeeIndex != nil && atts[i].AttestantionData().ValidatorIndex() != *committeeIndex {
+			continue
+		}
+		ret = append(ret, atts[i])
+	}
+
+	return newBeaconResponse(ret), nil
+}
+
+func (a *ApiHandler) PostEthV1BeaconPoolVoluntaryExits(w http.ResponseWriter, r *http.Request) {
+	req := cltypes.SignedVoluntaryExit{}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	if err := a.forkchoiceStore.OnVoluntaryExit(&req, false); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	// Broadcast to gossip
+	if a.sentinel != nil {
+		encodedSSZ, err := req.EncodeSSZ(nil)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		if _, err := a.sentinel.PublishGossip(r.Context(), &sentinel.GossipData{
+			Data: encodedSSZ,
+			Name: gossip.TopicNameVoluntaryExit,
+		}); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		a.operationsPool.VoluntaryExistsPool.Insert(req.VoluntaryExit.ValidatorIndex, &req)
+	}
+	// Only write 200
+	w.WriteHeader(http.StatusOK)
+}
+
+func (a *ApiHandler) PostEthV1BeaconPoolAttesterSlashings(w http.ResponseWriter, r *http.Request) {
+	req := cltypes.NewAttesterSlashing()
+	if err := json.NewDecoder(r.Body).Decode(req); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	if err := a.forkchoiceStore.OnAttesterSlashing(req, false); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	// Broadcast to gossip
+	if a.sentinel != nil {
+		encodedSSZ, err := req.EncodeSSZ(nil)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		if _, err := a.sentinel.PublishGossip(r.Context(), &sentinel.GossipData{
+			Data: encodedSSZ,
+			Name: gossip.TopicNameAttesterSlashing,
+		}); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+	}
+	// Only write 200
+	w.WriteHeader(http.StatusOK)
+}
+
+func (a *ApiHandler) PostEthV1BeaconPoolProposerSlashings(w http.ResponseWriter, r *http.Request) {
+	req := cltypes.ProposerSlashing{}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	if err := a.forkchoiceStore.OnProposerSlashing(&req, false); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	// Broadcast to gossip
+	if a.sentinel != nil {
+		encodedSSZ, err := req.EncodeSSZ(nil)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		if _, err := a.sentinel.PublishGossip(r.Context(), &sentinel.GossipData{
+			Data: encodedSSZ,
+			Name: gossip.TopicNameProposerSlashing,
+		}); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+	}
+	// Only write 200
+	w.WriteHeader(http.StatusOK)
+}
+
+type poolingFailure struct {
+	Index   int    `json:"index"`
+	Message string `json:"message"`
+}
+
+type poolingError struct {
+	Code     int              `json:"code"`
+	Message  string           `json:"message"`
+	Failures []poolingFailure `json:"failures"`
+}
+
+func (a *ApiHandler) PostEthV1BeaconPoolBlsToExecutionChanges(w http.ResponseWriter, r *http.Request) {
+	req := []*cltypes.SignedBLSToExecutionChange{}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	failures := []poolingFailure{}
+	for _, v := range req {
+		if err := a.forkchoiceStore.OnBlsToExecutionChange(v, false); err != nil {
+			failures = append(failures, poolingFailure{Index: len(failures), Message: err.Error()})
+			continue
+		}
+		// Broadcast to gossip
+		if a.sentinel != nil {
+			encodedSSZ, err := v.EncodeSSZ(nil)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			if _, err := a.sentinel.PublishGossip(r.Context(), &sentinel.GossipData{
+				Data: encodedSSZ,
+				Name: gossip.TopicNameBlsToExecutionChange,
+			}); err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+		}
+	}
+
+	if len(failures) > 0 {
+		w.WriteHeader(http.StatusBadRequest)
+		json.NewEncoder(w).Encode(poolingError{Code: http.StatusBadRequest, Message: "some failures", Failures: failures})
+		return
+	}
+	// Only write 200
+	w.WriteHeader(http.StatusOK)
 }