[r3.8] cl/phase1/network/registry: let one peer forward a slot's aggregates at once (#25250)

Cherry-pick of #25236 to release/3.8.

The `beacon_aggregate_and_proof` topic allowed 8 messages per second per
peer, with a burst of 16. Aggregates for a slot arrive in a burst, and
an aggregate the limiter ignored was also dropped from every other peer
because gossipsub had already marked it as seen, so the node lost whole
committees. One peer may now forward a whole slot's expected aggregates
in one burst (`MaxCommitteesPerSlot * TargetAggregatorsPerCommittee`),
refilled over the slot. Fixes #25235 on release/3.8.

Clean cherry-pick of 2cedbab. `go test -race
./cl/phase1/network/registry/` passes; `make lint` is clean twice; `make
erigon integration` builds. The same change has run on a
release/3.8-based mainnet node since 2026-10-22 (see #25236).

Author: lystopad

cl/phase1/network/registry/register.go

@@ -50,7 +50,7 @@ func RegisterGossipServices(
 	// register services
 	add(gossip.RegisterGossipService(gm, blockService, withRateLimiterByPeer(1, 2)))
 	add(gossip.RegisterGossipService(gm, syncContributionService, waitReady, withRateLimiterByPeer(8, 16)))
-	add(gossip.RegisterGossipService(gm, aggregateAndProofService, waitReady, withRateLimiterByPeer(8, 16)))
+	add(gossip.RegisterGossipService(gm, aggregateAndProofService, waitReady, withAggregateAndProofRateLimiter(beaconCfg)))
 	add(gossip.RegisterGossipService(gm, syncCommitteeMessagesService, waitReady, withRateLimiterByPeer(8, 16)))
 	add(gossip.RegisterGossipService(gm, attesterSlashingService, waitReady, withRateLimiterByPeer(2, 8)))
 	add(gossip.RegisterGossipService(gm, voluntaryExitService, waitReady, withRateLimiterByPeer(2, 8)))
@@ -107,6 +107,15 @@ func withPayloadAttestationRateLimiter() gossip.ConditionFunc {
 	return withRateLimiterByPeer(float64(clparams.MaxPtcSize), int(clparams.MaxPtcSize))
 }
 
+// withAggregateAndProofRateLimiter lets one peer forward every aggregate expected in a slot at once
+// (TARGET_AGGREGATORS_PER_COMMITTEE for each committee), refilled over the slot. Aggregates are
+// published at the same point in the slot, and one ignored here is also dropped from every other
+// peer, because gossipsub has already marked it as seen.
+func withAggregateAndProofRateLimiter(beaconCfg *clparams.BeaconChainConfig) gossip.ConditionFunc {
+	perSlot := beaconCfg.MaxCommitteesPerSlot * beaconCfg.TargetAggregatorsPerCommittee
+	return withRateLimiterByPeer(float64(perSlot)/float64(beaconCfg.SecondsPerSlot), int(perSlot))
+}
+
 // withProposerPreferencesRateLimiter lets one peer relay one preference for every slot of the
 // accepted lookahead epochs at once: as with payload attestations, an ignored preference cannot be
 // redelivered by another peer.

cl/phase1/network/registry/register_test.go

@@ -59,3 +59,26 @@ func TestProposerPreferencesRateLimiterAdmitsLookaheadWindowFromOnePeer(t *testi
 		})
 	}
 }
+
+// Aggregates for a slot are published at one point in the slot, so a single peer can forward
+// every committee's aggregators in one burst.
+func TestAggregateAndProofRateLimiterAdmitsSlotBurstFromOnePeer(t *testing.T) {
+	minimal := clparams.MainnetBeaconConfig
+	clparams.ApplyMinimalPreset(&minimal)
+	for _, test := range []struct {
+		name  string
+		cfg   clparams.BeaconChainConfig
+		burst int
+	}{
+		{name: "mainnet", cfg: clparams.MainnetBeaconConfig, burst: 1024},
+		{name: "minimal", cfg: minimal, burst: 64},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			condition := withAggregateAndProofRateLimiter(&test.cfg)
+			pid := peer.ID("aggregate-peer")
+			for i := range test.burst {
+				require.True(t, condition(pid, nil, clparams.FuluVersion), "aggregate %d", i)
+			}
+		})
+	}
+}