Merge branch 'main' into alex/mdbx_lag_34

Author: AskAlexSharov

cl/beacon/handler/block_production.go

@@ -47,6 +47,7 @@ import (
 	"github.com/erigontech/erigon/cl/gossip"
 	"github.com/erigontech/erigon/cl/persistence/beacon_indicies"
 	"github.com/erigontech/erigon/cl/phase1/core/state"
+	"github.com/erigontech/erigon/cl/phase1/network/subnets"
 	"github.com/erigontech/erigon/cl/transition"
 	"github.com/erigontech/erigon/cl/transition/impl/eth2"
 	"github.com/erigontech/erigon/cl/transition/machine"
@@ -143,8 +144,13 @@ func (a *ApiHandler) GetEthV1ValidatorAttestationData(
 	}
 
 	defer func() {
+		epoch := *slot / a.beaconChainCfg.SlotsPerEpoch
+		committeesPerSlot := a.syncedData.CommitteeCount(epoch)
+		subnet := subnets.ComputeSubnetForAttestation(
+			committeesPerSlot, *slot, *committeeIndex,
+			a.beaconChainCfg.SlotsPerEpoch, 64)
 		a.logger.Debug("Produced Attestation", "slot", *slot,
-			"committee_index", *committeeIndex, "cached", ok, "beacon_block_root",
+			"committee_index", *committeeIndex, "subnet", subnet, "cached", ok, "beacon_block_root",
 			attestationData.BeaconBlockRoot, "duration", time.Since(start))
 	}()
 

cl/p2p/mock_services/p2p_manager_mock.go

@@ -101,13 +101,22 @@ func (g *GossipManager) Close() error {
 }
 
 func (g *GossipManager) newPubsubValidator(service serviceintf.Service[any], conditions ...ConditionFunc) pubsub.ValidatorEx {
+	var selfID peer.ID
+	if h := g.p2p.Host(); h != nil {
+		selfID = h.ID()
+	}
 	return func(ctx context.Context, pid peer.ID, msg *pubsub.Message) (result pubsub.ValidationResult) {
 		defer func() {
 			if r := recover(); r != nil {
 				log.Error("[GossipManager] panic in validator, rejecting message", "err", r, "topic", msg.GetTopic())
 				result = pubsub.ValidationReject
 			}
 		}()
+		// Skip validation for self-published messages: they were already validated
+		// by ProcessMessage before Publish was called.
+		if selfID != "" && pid == selfID {
+			return pubsub.ValidationAccept
+		}
 		curVersion := g.beaconConfig.GetCurrentStateVersion(g.ethClock.GetCurrentEpoch())
 		// parse the topic and subnet
 		topic := msg.GetTopic()
@@ -252,8 +261,18 @@ func (g *GossipManager) Publish(ctx context.Context, name string, data []byte) e
 	if topicHandle == nil {
 		return fmt.Errorf("topic not found: %s", topic)
 	}
+	// Log peer count for attestation topics to help diagnose propagation issues
+	if gossip.IsTopicBeaconAttestation(name) {
+		peerCount := len(g.p2p.Pubsub().ListPeers(topic))
+		if peerCount == 0 {
+			log.Warn("[Gossip] Publishing attestation with NO peers on subnet", "topic", name, "peerCount", peerCount)
+		} else if peerCount < 3 {
+			log.Debug("[Gossip] Publishing attestation with low peer count", "topic", name, "peerCount", peerCount)
+		}
+	}
 	// Note: before publishing the message to the network, Publish() internally runs the validator function.
-	return topicHandle.topic.Publish(ctx, compressedData, pubsub.WithReadiness(pubsub.MinTopicSize(1)))
+	// Removed MinTopicSize(1) - don't fail if no peers on subnet, message will propagate when peers join
+	return topicHandle.topic.Publish(ctx, compressedData)
 }
 
 func (g *GossipManager) goCheckForkAndResubscribe(ctx context.Context) {

cl/phase1/network/gossip/gossip_manager.go

@@ -247,7 +247,7 @@ func (a *aggregateAndProofServiceImpl) ProcessMessage(
 			index: aggregateAndProof.SignedAggregateAndProof.Message.AggregatorIndex,
 		}
 		if a.seenAggreatorIndexes.Contains(seenIndex) {
-			return nil
+			return fmt.Errorf("%w: aggregator already seen", ErrIgnore)
 		}
 
 		committee, err := headState.GetBeaconCommitee(slot, committeeIndex)

cl/phase1/network/services/aggregate_and_proof_service.go

@@ -263,7 +263,7 @@ func (s *attestationService) ProcessMessage(ctx context.Context, subnet *uint64,
 		// mark the validator as seen
 		epochLastTime, ok := s.validatorAttestationSeen.Get(vIndex)
 		if ok && epochLastTime == targetEpoch {
-			return nil
+			return fmt.Errorf("validator already seen in target epoch %w", ErrIgnore)
 		}
 		s.validatorAttestationSeen.Add(vIndex, targetEpoch)