First round of fixes in making gossip publishing good for the validator: See comment

cl/aggregation/pool_impl.go

@@ -71,7 +71,7 @@ func (p *aggregationPoolImpl) AddAttestation(inAtt *solid.Attestation) error {
 	if err != nil {
 		return err
 	}
-	if len(merged) > 96 {
+	if len(merged) != 96 {
 		return fmt.Errorf("merged signature is too long")
 	}
 	var mergedSig [96]byte

cl/beacon/handler/handler.go

@@ -82,6 +82,7 @@ type ApiHandler struct {
 	syncCommitteeMessagesService     services.SyncCommitteeMessagesService
 	syncContributionAndProofsService services.SyncContributionService
 	aggregateAndProofsService        services.AggregateAndProofService
+	attestationService               services.AttestationService
 }
 
 func NewApiHandler(
@@ -110,6 +111,7 @@ func NewApiHandler(
 	syncCommitteeMessagesService services.SyncCommitteeMessagesService,
 	syncContributionAndProofs services.SyncContributionService,
 	aggregateAndProofs services.AggregateAndProofService,
+	attestationService services.AttestationService,
 ) *ApiHandler {
 	blobBundles, err := lru.New[common.Bytes48, BlobBundle]("blobs", maxBlobBundleCacheSize)
 	if err != nil {
@@ -146,6 +148,7 @@ func NewApiHandler(
 		syncCommitteeMessagesService:     syncCommitteeMessagesService,
 		syncContributionAndProofsService: syncContributionAndProofs,
 		aggregateAndProofsService:        aggregateAndProofs,
+		attestationService:               attestationService,
 	}
 }
 

cl/beacon/handler/pool.go

@@ -3,6 +3,7 @@ package handler
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/sentinel"
@@ -64,9 +65,20 @@ func (a *ApiHandler) PostEthV1BeaconPoolAttestations(w http.ResponseWriter, r *h
 		beaconhttp.NewEndpointError(http.StatusBadRequest, err).WriteTo(w)
 		return
 	}
+	headState := a.syncedData.HeadState()
+	if headState == nil {
+		beaconhttp.NewEndpointError(http.StatusServiceUnavailable, errors.New("head state not available")).WriteTo(w)
+		return
+	}
 	failures := []poolingFailure{}
 	for i, attestation := range req {
-		if err := a.forkchoiceStore.ValidateOnAttestation(attestation); err != nil {
+		var (
+			slot                  = attestation.AttestantionData().Slot()
+			cIndex                = attestation.AttestantionData().CommitteeIndex()
+			committeeCountPerSlot = headState.CommitteeCount(slot / uint64(a.beaconChainCfg.SlotsPerEpoch))
+			subnet                = subnets.ComputeSubnetForAttestation(committeeCountPerSlot, slot, cIndex, a.beaconChainCfg.SlotsPerEpoch, a.netConfig.AttestationSubnetCount)
+		)
+		if err := a.attestationService.ProcessMessage(r.Context(), &subnet, attestation); err != nil {
 			failures = append(failures, poolingFailure{
 				Index:   i,
 				Message: err.Error(),
@@ -81,18 +93,13 @@ func (a *ApiHandler) PostEthV1BeaconPoolAttestations(w http.ResponseWriter, r *h
 			continue
 		}
 		if a.sentinel != nil {
-			// broadcast
-			var (
-				slot                  = attestation.AttestantionData().Slot()
-				cIndex                = attestation.AttestantionData().CommitteeIndex()
-				committeeCountPerSlot = subnets.ComputeCommitteeCountPerSlot(a.syncedData.HeadState(), slot, a.beaconChainCfg.SlotsPerEpoch)
-				subnet                = subnets.ComputeSubnetForAttestation(committeeCountPerSlot, slot, cIndex, a.beaconChainCfg.SlotsPerEpoch, a.netConfig.AttestationSubnetCount)
-			)
+
 			encodedSSZ, err := attestation.EncodeSSZ(nil)
 			if err != nil {
 				beaconhttp.NewEndpointError(http.StatusInternalServerError, err).WriteTo(w)
 				return
 			}
+			fmt.Println(subnet)
 			if _, err := a.sentinel.PublishGossip(r.Context(), &sentinel.GossipData{
 				Data:     encodedSSZ,
 				Name:     gossip.TopicNameBeaconAttestation(subnet),

cl/phase1/network/services/attestation_service.go

@@ -15,8 +15,7 @@ import (
 )
 
 var (
-	computeCommitteeCountPerSlot = subnets.ComputeCommitteeCountPerSlot
-	computeSubnetForAttestation  = subnets.ComputeSubnetForAttestation
+	computeSubnetForAttestation = subnets.ComputeSubnetForAttestation
 )
 
 type attestationService struct {
@@ -55,14 +54,19 @@ func (s *attestationService) ProcessMessage(ctx context.Context, subnet *uint64,
 		root           = att.AttestantionData().BeaconBlockRoot()
 		slot           = att.AttestantionData().Slot()
 		committeeIndex = att.AttestantionData().CommitteeIndex()
-		epoch          = att.AttestantionData().Target().Epoch()
-		bits           = att.AggregationBits()
+		targetEpoch    = att.AttestantionData().Target().Epoch()
 	)
+	headState := s.syncedDataManager.HeadState()
+	if headState == nil {
+		return ErrIgnore
+	}
+
 	// [REJECT] The committee index is within the expected range
-	committeeCount := computeCommitteeCountPerSlot(s.syncedDataManager.HeadState(), slot, s.beaconCfg.SlotsPerEpoch)
+	committeeCount := headState.CommitteeCount(att.AttestantionData().Target().Epoch())
 	if committeeIndex >= committeeCount {
 		return fmt.Errorf("committee index out of range")
 	}
+	fmt.Println("received", *subnet)
 	// [REJECT] The attestation is for the correct subnet -- i.e. compute_subnet_for_attestation(committees_per_slot, attestation.data.slot, index) == subnet_id
 	subnetId := computeSubnetForAttestation(committeeCount, slot, committeeIndex, s.beaconCfg.SlotsPerEpoch, s.netCfg.AttestationSubnetCount)
 	if subnet == nil || subnetId != *subnet {
@@ -75,33 +79,37 @@ func (s *attestationService) ProcessMessage(ctx context.Context, subnet *uint64,
 		return fmt.Errorf("not in propagation range %w", ErrIgnore)
 	}
 	// [REJECT] The attestation's epoch matches its target -- i.e. attestation.data.target.epoch == compute_epoch_at_slot(attestation.data.slot)
-	if epoch != slot/s.beaconCfg.SlotsPerEpoch {
+	if targetEpoch != slot/s.beaconCfg.SlotsPerEpoch {
 		return fmt.Errorf("epoch mismatch")
 	}
+	// [REJECT] The number of aggregation bits matches the committee size -- i.e. len(aggregation_bits) == len(get_beacon_committee(state, attestation.data.slot, index)).
+	beaconCommitte, err := headState.GetBeaconCommitee(slot, committeeIndex)
+	if err != nil {
+		return err
+	}
+	expectedAggregationBitsLength := (len(beaconCommitte) + 7) / 8
+	if len(att.AggregationBits()) < expectedAggregationBitsLength {
+		return fmt.Errorf("aggregation bits count mismatch: %d != %d", len(att.AggregationBits()), expectedAggregationBitsLength)
+	}
+
+	bits := att.AggregationBits()
 	//[REJECT] The attestation is unaggregated -- that is, it has exactly one participating validator (len([bit for bit in aggregation_bits if bit]) == 1, i.e. exactly 1 bit is set).
-	emptyCount := 0
-	onBitIndex := 0
+	setBits := 0
+	onBitIndex := 0 // Aggregationbits is []byte, so we need to iterate over all bits.
 	for i := 0; i < len(bits); i++ {
-		if bits[i] == 0 {
-			emptyCount++
-		} else if bits[i]&(bits[i]-1) != 0 {
-			return fmt.Errorf("more than one bit set")
-		} else {
-			// find the onBit bit index
-			for onBit := uint(0); onBit < 8; onBit++ {
-				if bits[i]&(1<<onBit) != 0 {
-					onBitIndex = i*8 + int(onBit)
-					break
-				}
+		for j := 0; j < 8; j++ {
+			if bits[i]&(1<<uint(j)) != 0 {
+				setBits++
+				onBitIndex = i*8 + j
 			}
 		}
 	}
-	if emptyCount != len(bits)-1 {
-		return fmt.Errorf("no bit set")
+
+	if setBits == 0 {
+		return ErrIgnore
 	}
-	// [REJECT] The number of aggregation bits matches the committee size -- i.e. len(aggregation_bits) == len(get_beacon_committee(state, attestation.data.slot, index)).
-	if len(bits)*8 < int(committeeCount) {
-		return fmt.Errorf("aggregation bits count mismatch")
+	if setBits != 1 {
+		return fmt.Errorf("attestation does not have exactly one participating validator")
 	}
 
 	// [IGNORE] There has been no other valid attestation seen on an attestation subnet that has an identical attestation.data.target.epoch and participating validator index.
@@ -117,15 +125,15 @@ func (s *attestationService) ProcessMessage(ctx context.Context, subnet *uint64,
 		vIndex := committees[onBitIndex]
 		s.validatorAttSeenLock.Lock()
 		defer s.validatorAttSeenLock.Unlock()
-		if _, ok := s.validatorAttestationSeen[epoch]; !ok {
-			s.validatorAttestationSeen[epoch] = make(map[uint64]struct{})
+		if _, ok := s.validatorAttestationSeen[targetEpoch]; !ok {
+			s.validatorAttestationSeen[targetEpoch] = make(map[uint64]struct{})
 		}
-		if _, ok := s.validatorAttestationSeen[epoch][vIndex]; ok {
+		if _, ok := s.validatorAttestationSeen[targetEpoch][vIndex]; ok {
 			return fmt.Errorf("validator already seen in target epoch %w", ErrIgnore)
 		}
-		s.validatorAttestationSeen[epoch][vIndex] = struct{}{}
+		s.validatorAttestationSeen[targetEpoch][vIndex] = struct{}{}
 		// always check and delete previous epoch if it exists
-		delete(s.validatorAttestationSeen, epoch-1)
+		delete(s.validatorAttestationSeen, targetEpoch-1)
 		return nil
 	}(); err != nil {
 		return err
@@ -139,7 +147,7 @@ func (s *attestationService) ProcessMessage(ctx context.Context, subnet *uint64,
 
 	// [REJECT] The attestation's target block is an ancestor of the block named in the LMD vote -- i.e.
 	// get_checkpoint_block(store, attestation.data.beacon_block_root, attestation.data.target.epoch) == attestation.data.target.root
-	startSlotAtEpoch := epoch * s.beaconCfg.SlotsPerEpoch
+	startSlotAtEpoch := targetEpoch * s.beaconCfg.SlotsPerEpoch
 	if s.forkchoiceStore.Ancestor(root, startSlotAtEpoch) != att.AttestantionData().Target().BlockRoot() {
 		return fmt.Errorf("invalid target block")
 	}

cl/phase1/network/subnets/subnets.go

@@ -63,8 +63,3 @@ func ComputeSubnetForAttestation(committeePerSlot, slot, committeeIndex, slotsPe
 	committeesSinceEpochStart := committeePerSlot * slotsSinceEpochStart
 	return (committeesSinceEpochStart + committeeIndex) % attSubnetCount
 }
-
-func ComputeCommitteeCountPerSlot(s *state.CachingBeaconState, slot uint64, slotsPerEpoch uint64) uint64 {
-	epoch := slot / slotsPerEpoch
-	return s.CommitteeCount(epoch)
-}

cl/validator/attestation_producer/attestation_producer.go

@@ -74,7 +74,7 @@ func (ap *attestationProducer) ProduceAndCacheAttestationData(baseState *state.C
 	}
 
 	targetEpoch := state.Epoch(baseState)
-	epochStartTargetSlot := (targetEpoch * ap.beaconCfg.SlotsPerEpoch) - (ap.beaconCfg.SlotsPerEpoch - 1)
+	epochStartTargetSlot := targetEpoch * ap.beaconCfg.SlotsPerEpoch
 	var targetRoot libcommon.Hash
 	if epochStartTargetSlot == baseState.Slot() {
 		targetRoot = baseStateBlockRoot

cl/validator/committee_subscription/committee_subscription.go

@@ -81,8 +81,12 @@ func (c *CommitteeSubscribeMgmt) AddAttestationSubscription(ctx context.Context,
 		cIndex = p.CommitteeIndex
 		vIndex = p.ValidatorIndex
 	)
+	headState := c.syncedData.HeadState()
+	if headState == nil {
+		return fmt.Errorf("head state not available")
+	}
 
-	commiteePerSlot := subnets.ComputeCommitteeCountPerSlot(c.syncedData.HeadState(), slot, c.beaconConfig.SlotsPerEpoch)
+	commiteePerSlot := headState.CommitteeCount(p.Slot / c.beaconConfig.SlotsPerEpoch)
 	subnetId := subnets.ComputeSubnetForAttestation(commiteePerSlot, slot, cIndex, c.beaconConfig.SlotsPerEpoch, c.netConfig.AttestationSubnetCount)
 	// add validator to subscription
 	c.validatorSubsMutex.Lock()

cmd/caplin/caplin1/run.go

@@ -290,6 +290,7 @@ func RunCaplinPhase1(ctx context.Context, engine execution_client.ExecutionEngin
 			syncCommitteeMessagesService,
 			syncContributionService,
 			aggregateAndProofService,
+			attestationService,
 		)
 		go beacon.ListenAndServe(&beacon.LayeredBeaconHandler{
 			ArchiveApi: apiHandler,