Skip to content

feat: use static subscribe instead of subscribeWithParameters for blob sidecar topics #15895

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: develop
Choose a base branch
from
Open

feat: use static subscribe instead of subscribeWithParameters for blob sidecar topics #15895

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8fb539d
use static subscribe for blob sidecar topics
aarshkshah1992 Oct 20, 2025
aebfd8d
Merge branch 'develop' into feat/use-subscribe-for-blobs
aarshkshah1992 Oct 20, 2025
e7be659
add changelog
aarshkshah1992 Oct 20, 2025
9cfddb8
fix changelog CI
aarshkshah1992 Oct 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 57 additions & 56 deletions beacon-chain/sync/subscriber.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,11 +79,6 @@ func (p subscribeParameters) logFields() logrus.Fields {
}
}

// fullTopic is the fully qualified topic string, given to gossipsub.
func (p subscribeParameters) fullTopic(subnet uint64, suffix string) string {
return fmt.Sprintf(p.topicFormat, p.nse.ForkDigest, subnet) + suffix
}

// subnetTracker keeps track of which subnets we are subscribed to, out of the set of
// possible subnets described by a `subscribeParameters`.
type subnetTracker struct {
Expand Down Expand Up @@ -210,19 +205,20 @@ func (s *Service) registerSubscribers(nse params.NetworkScheduleEntry) bool {
return false
}
s.spawn(func() {
s.subscribe(p2p.BlockSubnetTopicFormat, s.validateBeaconBlockPubSub, s.beaconBlockSubscriber, nse)
s.subscribe(p2p.BlockSubnetTopicFormat, s.buildTopicWithoutSubnet(p2p.BlockSubnetTopicFormat,
nse.ForkDigest), s.validateBeaconBlockPubSub, s.beaconBlockSubscriber, nse)
})
s.spawn(func() {
s.subscribe(p2p.AggregateAndProofSubnetTopicFormat, s.validateAggregateAndProof, s.beaconAggregateProofSubscriber, nse)
s.subscribe(p2p.AggregateAndProofSubnetTopicFormat, s.buildTopicWithoutSubnet(p2p.AggregateAndProofSubnetTopicFormat, nse.ForkDigest), s.validateAggregateAndProof, s.beaconAggregateProofSubscriber, nse)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

s.buildTopicWithoutSubnet could be called and the result extracted in a variable to avoid extra long lines of code.

})
s.spawn(func() {
s.subscribe(p2p.ExitSubnetTopicFormat, s.validateVoluntaryExit, s.voluntaryExitSubscriber, nse)
s.subscribe(p2p.ExitSubnetTopicFormat, s.buildTopicWithoutSubnet(p2p.ExitSubnetTopicFormat, nse.ForkDigest), s.validateVoluntaryExit, s.voluntaryExitSubscriber, nse)
})
s.spawn(func() {
s.subscribe(p2p.ProposerSlashingSubnetTopicFormat, s.validateProposerSlashing, s.proposerSlashingSubscriber, nse)
s.subscribe(p2p.ProposerSlashingSubnetTopicFormat, s.buildTopicWithoutSubnet(p2p.ProposerSlashingSubnetTopicFormat, nse.ForkDigest), s.validateProposerSlashing, s.proposerSlashingSubscriber, nse)
})
s.spawn(func() {
s.subscribe(p2p.AttesterSlashingSubnetTopicFormat, s.validateAttesterSlashing, s.attesterSlashingSubscriber, nse)
s.subscribe(p2p.AttesterSlashingSubnetTopicFormat, s.buildTopicWithoutSubnet(p2p.AttesterSlashingSubnetTopicFormat, nse.ForkDigest), s.validateAttesterSlashing, s.attesterSlashingSubscriber, nse)
})
s.spawn(func() {
s.subscribeWithParameters(subscribeParameters{
Expand All @@ -240,6 +236,7 @@ func (s *Service) registerSubscribers(nse params.NetworkScheduleEntry) bool {
s.spawn(func() {
s.subscribe(
p2p.SyncContributionAndProofSubnetTopicFormat,
s.buildTopicWithoutSubnet(p2p.SyncContributionAndProofSubnetTopicFormat, nse.ForkDigest),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • All the parameters used in s.buildTopicWihoutSubnet are also parameters of subscribe.
  • All calls of s.subscribe also use s.builTopicWithoutSubnet

==> s.buildTopicWithoutSubnet should be directly called in s.subscribe.

s.validateSyncContributionAndProof,
s.syncContributionAndProofSubscriber,
nse,
Expand All @@ -259,6 +256,7 @@ func (s *Service) registerSubscribers(nse params.NetworkScheduleEntry) bool {
s.spawn(func() {
s.subscribe(
p2p.LightClientOptimisticUpdateTopicFormat,
s.buildTopicWithoutSubnet(p2p.LightClientOptimisticUpdateTopicFormat, nse.ForkDigest),
s.validateLightClientOptimisticUpdate,
noopHandler,
nse,
Expand All @@ -267,6 +265,7 @@ func (s *Service) registerSubscribers(nse params.NetworkScheduleEntry) bool {
s.spawn(func() {
s.subscribe(
p2p.LightClientFinalityUpdateTopicFormat,
s.buildTopicWithoutSubnet(p2p.LightClientFinalityUpdateTopicFormat, nse.ForkDigest),
s.validateLightClientFinalityUpdate,
noopHandler,
nse,
Expand All @@ -280,6 +279,7 @@ func (s *Service) registerSubscribers(nse params.NetworkScheduleEntry) bool {
s.spawn(func() {
s.subscribe(
p2p.BlsToExecutionChangeSubnetTopicFormat,
s.buildTopicWithoutSubnet(p2p.BlsToExecutionChangeSubnetTopicFormat, nse.ForkDigest),
s.validateBlsToExecutionChange,
s.blsToExecutionChangeSubscriber,
nse,
Expand All @@ -289,32 +289,36 @@ func (s *Service) registerSubscribers(nse params.NetworkScheduleEntry) bool {

// New gossip topic in Deneb, removed in Electra
if params.BeaconConfig().DenebForkEpoch <= nse.Epoch && nse.Epoch < params.BeaconConfig().ElectraForkEpoch {
s.spawn(func() {
s.subscribeWithParameters(subscribeParameters{
topicFormat: p2p.BlobSubnetTopicFormat,
validate: s.validateBlob,
handle: s.blobSubscriber,
nse: nse,
getSubnetsToJoin: func(primitives.Slot) map[uint64]bool {
return mapFromCount(params.BeaconConfig().BlobsidecarSubnetCount)
},
for i := uint64(0); i < params.BeaconConfig().BlobsidecarSubnetCount; i++ {
subnet := i
s.spawn(func() {
topic := s.buildTopicWithSubnet(p2p.BlobSubnetTopicFormat, nse.ForkDigest, subnet)
s.subscribe(
p2p.BlobSubnetTopicFormat,
topic,
s.validateBlob,
s.blobSubscriber,
nse,
)
})
})
}
}

// New gossip topic in Electra, removed in Fulu
if params.BeaconConfig().ElectraForkEpoch <= nse.Epoch && nse.Epoch < params.BeaconConfig().FuluForkEpoch {
s.spawn(func() {
s.subscribeWithParameters(subscribeParameters{
topicFormat: p2p.BlobSubnetTopicFormat,
validate: s.validateBlob,
handle: s.blobSubscriber,
nse: nse,
getSubnetsToJoin: func(currentSlot primitives.Slot) map[uint64]bool {
return mapFromCount(params.BeaconConfig().BlobsidecarSubnetCountElectra)
},
for i := uint64(0); i < params.BeaconConfig().BlobsidecarSubnetCountElectra; i++ {
subnet := i
s.spawn(func() {
topic := s.buildTopicWithSubnet(p2p.BlobSubnetTopicFormat, nse.ForkDigest, subnet)
s.subscribe(
p2p.BlobSubnetTopicFormat,
topic,
s.validateBlob,
s.blobSubscriber,
nse,
)
})
})
}
}

// New gossip topic in Fulu.
Expand Down Expand Up @@ -349,27 +353,40 @@ func (s *Service) subscribeLogFields(topic string, nse params.NetworkScheduleEnt

// subscribe to a given topic with a given validator and subscription handler.
// The base protobuf message is used to initialize new messages for decoding.
func (s *Service) subscribe(topic string, validator wrappedVal, handle subHandler, nse params.NetworkScheduleEntry) {
func (s *Service) subscribe(topicFormat string, fullTopic string, validator wrappedVal, handle subHandler, nse params.NetworkScheduleEntry) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why using both topicFormat and fullTopic here?

It seems topicFormat is only used in logs and in p2p.GossipTopicMappings, where the result is only checked against nil.

if err := s.waitForInitialSync(s.ctx); err != nil {
log.WithFields(s.subscribeLogFields(topic, nse)).WithError(err).Debug("Context cancelled while waiting for initial sync, not subscribing to topic")
log.WithFields(s.subscribeLogFields(topicFormat, nse)).WithError(err).Debug("Context cancelled while waiting for initial sync, not subscribing to topic")
return
}
// Check if this subscribe request is still valid - we may have crossed another fork epoch while waiting for initial sync.
if s.subscriptionRequestExpired(nse) {
// If we are already past the next fork epoch, do not subscribe to this topic.
log.WithFields(s.subscribeLogFields(topic, nse)).Debug("Not subscribing to topic as we are already past the next fork epoch")
log.WithFields(s.subscribeLogFields(topicFormat, nse)).Debug("Not subscribing to topic as we are already past the next fork epoch")
return
}
base := p2p.GossipTopicMappings(topic, nse.Epoch)
base := p2p.GossipTopicMappings(topicFormat, nse.Epoch)
if base == nil {
// Impossible condition as it would mean topic does not exist.
panic(fmt.Sprintf("%s is not mapped to any message in GossipTopicMappings", topic)) // lint:nopanic -- Impossible condition.
panic(fmt.Sprintf("%s is not mapped to any message in GossipTopicMappings", topicFormat)) // lint:nopanic -- Impossible condition.
}
s.subscribeWithBase(fullTopic, validator, handle)
}

func (s *Service) buildTopicWithoutSubnet(topicFormat string, digest [4]byte) string {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Without taking tests into account, buildTopicWithoutSubnet is only used when using subscribe.
==> The call to buildTopicWithoutSubnet could be directly done in subscribe, and thus avoiding an extra parameter in subscribe.

if !strings.Contains(topicFormat, "%x") {
log.Error("Topic does not have appropriate formatter for digest")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we are here, the user have no idea of which subnet was wrong.
IMO, it's better to return an error here.

}
return (fmt.Sprintf(topicFormat, digest)) + s.cfg.p2p.Encoding().ProtocolSuffix()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why the extra parenthesis around fmt.Sprintf?

}

func (s *Service) buildTopicWithSubnet(topicFormat string, digest [4]byte, subnet uint64) string {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same comments as in buildTopicWithoutSubnet.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The names buildTopicWithSubnet and buildTopicWithoutSubnet are not so clear.
Maybe something like buildStaticSubnet and buildDynamicSubnet?

Also, please add a godoc with an example of suitable topic format, so we can understand what the function does without reading it.

if !strings.Contains(topicFormat, "%x") {
log.Error("Topic does not have appropriate formatter for digest")
}
s.subscribeWithBase(s.addDigestToTopic(topic, nse.ForkDigest), validator, handle)
return (fmt.Sprintf(topicFormat, digest, subnet)) + s.cfg.p2p.Encoding().ProtocolSuffix()
}

func (s *Service) subscribeWithBase(topic string, validator wrappedVal, handle subHandler) *pubsub.Subscription {
topic += s.cfg.p2p.Encoding().ProtocolSuffix()
log := log.WithField("topic", topic)

// Do not resubscribe already seen subscriptions.
Expand Down Expand Up @@ -532,7 +549,8 @@ func (s *Service) wrapAndReportValidation(topic string, v wrappedVal) (string, p
func (s *Service) pruneNotWanted(t *subnetTracker, wantedSubnets map[uint64]bool) {
for _, subnet := range t.unwanted(wantedSubnets) {
t.cancelSubscription(subnet)
s.unSubscribeFromTopic(t.fullTopic(subnet, s.cfg.p2p.Encoding().ProtocolSuffix()))
fullTopic := s.buildTopicWithSubnet(t.topicFormat, t.nse.ForkDigest, subnet)
s.unSubscribeFromTopic(fullTopic)
}
}

Expand Down Expand Up @@ -579,8 +597,7 @@ func (s *Service) trySubscribeSubnets(t *subnetTracker) {
subnetsToJoin := t.getSubnetsToJoin(s.cfg.clock.CurrentSlot())
s.pruneNotWanted(t, subnetsToJoin)
for _, subnet := range t.missing(subnetsToJoin) {
// TODO: subscribeWithBase appends the protocol suffix, other methods don't. Make this consistent.
topic := t.fullTopic(subnet, "")
topic := s.buildTopicWithSubnet(t.topicFormat, t.nse.ForkDigest, subnet)
t.track(subnet, s.subscribeWithBase(topic, t.validate, t.handle))
}
}
Expand Down Expand Up @@ -782,22 +799,6 @@ func (s *Service) filterNeededPeers(pids []peer.ID) []peer.ID {
return newPeers
}

// Add fork digest to topic.
func (*Service) addDigestToTopic(topic string, digest [4]byte) string {
if !strings.Contains(topic, "%x") {
log.Error("Topic does not have appropriate formatter for digest")
}
return fmt.Sprintf(topic, digest)
}

// Add the digest and index to subnet topic.
func (*Service) addDigestAndIndexToTopic(topic string, digest [4]byte, idx uint64) string {
if !strings.Contains(topic, "%x") {
log.Error("Topic does not have appropriate formatter for digest")
}
return fmt.Sprintf(topic, digest, idx)
}

func (s *Service) currentForkDigest() ([4]byte, error) {
return params.ForkDigest(s.cfg.clock.CurrentEpoch()), nil
}
Expand Down
26 changes: 15 additions & 11 deletions beacon-chain/sync/subscriber_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,8 @@ func TestSubscribe_ReceivesValidMessage(t *testing.T) {
var wg sync.WaitGroup
wg.Add(1)

r.subscribe(topic, r.noopValidator, func(_ context.Context, msg proto.Message) error {
fullTopic := r.buildTopicWithoutSubnet(topic, nse.ForkDigest)
r.subscribe(topic, fullTopic, r.noopValidator, func(_ context.Context, msg proto.Message) error {
m, ok := msg.(*pb.SignedVoluntaryExit)
assert.Equal(t, true, ok, "Object is not of type *pb.SignedVoluntaryExit")
if m.Exit == nil || m.Exit.Epoch != 55 {
Expand Down Expand Up @@ -110,12 +111,12 @@ func TestSubscribe_UnsubscribeTopic(t *testing.T) {
p2pService.Digest = nse.ForkDigest
topic := "/eth2/%x/voluntary_exit"

r.subscribe(topic, r.noopValidator, func(_ context.Context, msg proto.Message) error {
fullTopic := r.buildTopicWithoutSubnet(topic, nse.ForkDigest)
r.subscribe(topic, fullTopic, r.noopValidator, func(_ context.Context, msg proto.Message) error {
return nil
}, nse)
r.markForChainStart()

fullTopic := fmt.Sprintf(topic, p2pService.Digest) + p2pService.Encoding().ProtocolSuffix()
assert.Equal(t, true, r.subHandler.topicExists(fullTopic))
topics := p2pService.PubSub().GetTopics()
assert.Equal(t, fullTopic, topics[0])
Expand Down Expand Up @@ -162,7 +163,8 @@ func TestSubscribe_ReceivesAttesterSlashing(t *testing.T) {
wg.Add(1)
nse := params.GetNetworkScheduleEntry(r.cfg.clock.CurrentEpoch())
p2pService.Digest = nse.ForkDigest
r.subscribe(topic, r.noopValidator, func(ctx context.Context, msg proto.Message) error {
fullTopic := r.buildTopicWithoutSubnet(topic, nse.ForkDigest)
r.subscribe(topic, fullTopic, r.noopValidator, func(ctx context.Context, msg proto.Message) error {
require.NoError(t, r.attesterSlashingSubscriber(ctx, msg))
wg.Done()
return nil
Expand Down Expand Up @@ -217,7 +219,8 @@ func TestSubscribe_ReceivesProposerSlashing(t *testing.T) {
params.OverrideBeaconConfig(params.MainnetConfig())
nse := params.GetNetworkScheduleEntry(r.cfg.clock.CurrentEpoch())
p2pService.Digest = nse.ForkDigest
r.subscribe(topic, r.noopValidator, func(ctx context.Context, msg proto.Message) error {
fullTopic := r.buildTopicWithoutSubnet(topic, nse.ForkDigest)
r.subscribe(topic, fullTopic, r.noopValidator, func(ctx context.Context, msg proto.Message) error {
require.NoError(t, r.proposerSlashingSubscriber(ctx, msg))
wg.Done()
return nil
Expand Down Expand Up @@ -266,7 +269,8 @@ func TestSubscribe_HandlesPanic(t *testing.T) {
var wg sync.WaitGroup
wg.Add(1)

r.subscribe(topic, r.noopValidator, func(_ context.Context, msg proto.Message) error {
fullTopic := r.buildTopicWithoutSubnet(topic, nse.ForkDigest)
r.subscribe(topic, fullTopic, r.noopValidator, func(_ context.Context, msg proto.Message) error {
defer wg.Done()
panic("bad")
}, nse)
Expand Down Expand Up @@ -305,7 +309,7 @@ func TestRevalidateSubscription_CorrectlyFormatsTopic(t *testing.T) {

// committee index 1
c1 := uint64(1)
fullTopic := params.fullTopic(c1, r.cfg.p2p.Encoding().ProtocolSuffix())
fullTopic := r.buildTopicWithSubnet(params.topicFormat, params.nse.ForkDigest, c1)
_, topVal := r.wrapAndReportValidation(fullTopic, r.noopValidator)
require.NoError(t, r.cfg.p2p.PubSub().RegisterTopicValidator(fullTopic, topVal))
sub1, err := r.cfg.p2p.SubscribeToTopic(fullTopic)
Expand All @@ -314,7 +318,7 @@ func TestRevalidateSubscription_CorrectlyFormatsTopic(t *testing.T) {

// committee index 2
c2 := uint64(2)
fullTopic = params.fullTopic(c2, r.cfg.p2p.Encoding().ProtocolSuffix())
fullTopic = r.buildTopicWithSubnet(params.topicFormat, params.nse.ForkDigest, c2)
_, topVal = r.wrapAndReportValidation(fullTopic, r.noopValidator)
err = r.cfg.p2p.PubSub().RegisterTopicValidator(fullTopic, topVal)
require.NoError(t, err)
Expand Down Expand Up @@ -484,11 +488,11 @@ func TestFilterSubnetPeers(t *testing.T) {
defer cache.SubnetIDs.EmptyAllCaches()
digest, err := r.currentForkDigest()
assert.NoError(t, err)
defaultTopic := "/eth2/%x/beacon_attestation_%d" + r.cfg.p2p.Encoding().ProtocolSuffix()
subnet10 := r.addDigestAndIndexToTopic(defaultTopic, digest, 10)
defaultTopic := "/eth2/%x/beacon_attestation_%d"
subnet10 := r.buildTopicWithSubnet(defaultTopic, digest, 10)
cache.SubnetIDs.AddAggregatorSubnetID(currSlot, 10)

subnet20 := r.addDigestAndIndexToTopic(defaultTopic, digest, 20)
subnet20 := r.buildTopicWithSubnet(defaultTopic, digest, 20)
cache.SubnetIDs.AddAttesterSubnetID(currSlot, 20)

p1 := createPeer(t, subnet10)
Expand Down
6 changes: 3 additions & 3 deletions beacon-chain/sync/sync_fuzz_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ func FuzzValidateBeaconBlockPubSub_Phase0(f *testing.F) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(msg)]
digest, err := r.currentForkDigest()
assert.NoError(f, err)
topic = r.addDigestToTopic(topic, digest)
topic = r.buildTopicWithoutSubnet(topic, digest)

f.Add("junk", []byte("junk"), buf.Bytes(), []byte(topic))
f.Fuzz(func(t *testing.T, pid string, from, data, topic []byte) {
Expand Down Expand Up @@ -166,7 +166,7 @@ func FuzzValidateBeaconBlockPubSub_Altair(f *testing.F) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(msg)]
digest, err := r.currentForkDigest()
assert.NoError(f, err)
topic = r.addDigestToTopic(topic, digest)
topic = r.buildTopicWithoutSubnet(topic, digest)

f.Add("junk", []byte("junk"), buf.Bytes(), []byte(topic))
f.Fuzz(func(t *testing.T, pid string, from, data, topic []byte) {
Expand Down Expand Up @@ -250,7 +250,7 @@ func FuzzValidateBeaconBlockPubSub_Bellatrix(f *testing.F) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(msg)]
digest, err := r.currentForkDigest()
assert.NoError(f, err)
topic = r.addDigestToTopic(topic, digest)
topic = r.buildTopicWithoutSubnet(topic, digest)

f.Add("junk", []byte("junk"), buf.Bytes(), []byte(topic))
f.Fuzz(func(t *testing.T, pid string, from, data, topic []byte) {
Expand Down
4 changes: 2 additions & 2 deletions beacon-chain/sync/validate_aggregate_proof_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -488,7 +488,7 @@ func TestValidateAggregateAndProof_CanValidate(t *testing.T) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(signedAggregateAndProof)]
d, err := r.currentForkDigest()
assert.NoError(t, err)
topic = r.addDigestToTopic(topic, d)
topic = r.buildTopicWithoutSubnet(topic, d)
msg := &pubsub.Message{
Message: &pubsubpb.Message{
Data: buf.Bytes(),
Expand Down Expand Up @@ -591,7 +591,7 @@ func TestVerifyIndexInCommittee_SeenAggregatorEpoch(t *testing.T) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(signedAggregateAndProof)]
d, err := r.currentForkDigest()
assert.NoError(t, err)
topic = r.addDigestToTopic(topic, d)
topic = r.buildTopicWithoutSubnet(topic, d)
msg := &pubsub.Message{
Message: &pubsubpb.Message{
Data: buf.Bytes(),
Expand Down
8 changes: 4 additions & 4 deletions beacon-chain/sync/validate_attester_slashing_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ func TestValidateAttesterSlashing_ValidSlashing(t *testing.T) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
d, err := r.currentForkDigest()
assert.NoError(t, err)
topic = r.addDigestToTopic(topic, d)
topic = r.buildTopicWithoutSubnet(topic, d)
msg := &pubsub.Message{
Message: &pubsubpb.Message{
Data: buf.Bytes(),
Expand Down Expand Up @@ -146,7 +146,7 @@ func TestValidateAttesterSlashing_ValidOldSlashing(t *testing.T) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
d, err := r.currentForkDigest()
assert.NoError(t, err)
topic = r.addDigestToTopic(topic, d)
topic = r.buildTopicWithoutSubnet(topic, d)
msg := &pubsub.Message{
Message: &pubsubpb.Message{
Data: buf.Bytes(),
Expand Down Expand Up @@ -191,7 +191,7 @@ func TestValidateAttesterSlashing_InvalidSlashing_WithdrawableEpoch(t *testing.T
topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
d, err := r.currentForkDigest()
assert.NoError(t, err)
topic = r.addDigestToTopic(topic, d)
topic = r.buildTopicWithoutSubnet(topic, d)
msg := &pubsub.Message{
Message: &pubsubpb.Message{
Data: buf.Bytes(),
Expand Down Expand Up @@ -240,7 +240,7 @@ func TestValidateAttesterSlashing_CanFilter(t *testing.T) {
topic := p2p.GossipTypeMapping[reflect.TypeOf(&ethpb.AttesterSlashing{})]
d, err := r.currentForkDigest()
assert.NoError(t, err)
topic = r.addDigestToTopic(topic, d)
topic = r.buildTopicWithoutSubnet(topic, d)
buf := new(bytes.Buffer)
_, err = p.Encoding().EncodeGossip(buf, &ethpb.AttesterSlashing{
Attestation_1: util.HydrateIndexedAttestation(&ethpb.IndexedAttestation{
Expand Down
Loading
Loading