diff --git a/comm.go b/comm.go index 234572af..b62242f7 100644 --- a/comm.go +++ b/comm.go @@ -24,6 +24,10 @@ func (p *PubSub) getHelloPacket() *RPC { subscriptions := make(map[string]bool) for t := range p.mySubs { + // don't announce fanout-only topics + if topic := p.myTopics[t]; topic != nil && topic.fanoutOnly { + continue + } subscriptions[t] = true } diff --git a/gossipsub_test.go b/gossipsub_test.go index b309abbe..fc13cc11 100644 --- a/gossipsub_test.go +++ b/gossipsub_test.go @@ -5405,3 +5405,121 @@ outer: } } } + +func TestGossipsubFanoutOnly(t *testing.T) { + // Test that a fanout-only topic can publish to the network but its + // subscriber only receives locally published messages. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + hosts := getDefaultHosts(t, 5) + psubs := getGossipsubs(ctx, hosts) + + topicID := "foobar" + + // hosts[0] joins with FanoutOnly - it should be able to publish but never + // join the mesh, so its subscriber should not receive remote messages. + fanoutTopic, err := psubs[0].Join(topicID, FanoutOnly()) + if err != nil { + t.Fatal(err) + } + + // The rest subscribe normally. + var normalSubs []*Subscription + for _, ps := range psubs[1:] { + sub, err := ps.Subscribe(topicID) + if err != nil { + t.Fatal(err) + } + normalSubs = append(normalSubs, sub) + } + + // Also subscribe on the fanout-only topic. Since it's fanout-only, this + // must not trigger a p2p subscription. + fanoutSub, err := fanoutTopic.Subscribe() + if err != nil { + t.Fatal(err) + } + + denseConnect(t, hosts) + + // Wait for heartbeats to build mesh. + time.Sleep(2 * time.Second) + + // Publish from the fanout-only node. Normal subscribers should receive it + // because the router uses fanout to forward the message. + fanoutMsg := []byte("from fanout-only node") + if err := fanoutTopic.Publish(ctx, fanoutMsg); err != nil { + t.Fatal(err) + } + + for i, sub := range normalSubs { + tctx, tcancel := context.WithTimeout(ctx, 5*time.Second) + got, err := sub.Next(tctx) + tcancel() + if err != nil { + t.Fatalf("normal sub %d did not receive fanout message: %v", i, err) + } + if !bytes.Equal(got.Data, fanoutMsg) { + t.Fatalf("normal sub %d got wrong message", i) + } + } + + // The fanout subscriber should also get the locally published message. + tctx, tcancel := context.WithTimeout(ctx, 5*time.Second) + got, err := fanoutSub.Next(tctx) + tcancel() + if err != nil { + t.Fatal("fanout subscriber did not receive locally published message:", err) + } + if !bytes.Equal(got.Data, fanoutMsg) { + t.Fatal("fanout subscriber got wrong message") + } + + // Now publish from a normal node. The fanout subscriber must NOT receive + // it because the fanout-only topic never joined the mesh. + remoteMsg := []byte("from normal node") + if err := psubs[1].Publish(topicID, remoteMsg); err != nil { + t.Fatal(err) + } + + // The other normal subscribers should receive it. + for i, sub := range normalSubs[1:] { + tctx, tcancel := context.WithTimeout(ctx, 5*time.Second) + got, err := sub.Next(tctx) + tcancel() + if err != nil { + t.Fatalf("normal sub %d did not receive remote message: %v", i+1, err) + } + if !bytes.Equal(got.Data, remoteMsg) { + t.Fatalf("normal sub %d got wrong message", i+1) + } + } + + // The fanout subscriber should NOT receive the remote message. + tctx, tcancel = context.WithTimeout(ctx, time.Second) + _, err = fanoutSub.Next(tctx) + tcancel() + if err == nil { + t.Fatal("fanout subscriber received a remote message but should not have") + } +} + +func TestGossipsubFanoutOnlyRelay(t *testing.T) { + // Test that Relay() returns an error on a fanout-only topic. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + hosts := getDefaultHosts(t, 1) + ps := getGossipsub(ctx, hosts[0]) + + topic, err := ps.Join("foobar", FanoutOnly()) + if err != nil { + t.Fatal(err) + } + + _, err = topic.Relay() + if !errors.Is(err, ErrFanoutOnlyTopic) { + t.Fatalf("expected ErrFanoutOnlyTopic, got: %v", err) + } +} diff --git a/pubsub.go b/pubsub.go index 2cf8e4bb..9d759cb5 100644 --- a/pubsub.go +++ b/pubsub.go @@ -1107,8 +1107,12 @@ func (p *PubSub) handleRemoveSubscription(sub *Subscription) { // stop announcing only if there are no more subs and relays if p.myRelays[sub.topic] == 0 { p.disc.StopAdvertise(sub.topic) - p.announce(sub.topic, false) - p.rt.Leave(sub.topic) + // skip mesh unsubscription for fanout-only topics since we never joined + topic := p.myTopics[sub.topic] + if topic == nil || !topic.fanoutOnly { + p.announce(sub.topic, false) + p.rt.Leave(sub.topic) + } } } } @@ -1124,8 +1128,13 @@ func (p *PubSub) handleAddSubscription(req *addSubReq) { // announce we want this topic if neither subs nor relays exist so far if len(subs) == 0 && p.myRelays[sub.topic] == 0 { p.disc.Advertise(sub.topic) - p.announce(sub.topic, true) - p.rt.Join(sub.topic) + // skip mesh subscription for fanout-only topics; + // discovery/advertising is still needed to find peers for fanout publishing + topic := p.myTopics[sub.topic] + if topic == nil || !topic.fanoutOnly { + p.announce(sub.topic, true) + p.rt.Join(sub.topic) + } } // make new if not there @@ -1605,6 +1614,17 @@ func SupportsPartialMessages() TopicOpt { type TopicOpt func(t *Topic) error +// FanoutOnly enforces fanout-only mode for the topic. In this mode, the node can publish +// messages to the topic but will never subscribe to the p2p mesh, even if Topic.Subscribe +// is called. Subscribers will only receive locally published messages via Topic.Publish. +// Calling Topic.Relay on a fanout-only topic will return ErrFanoutOnlyTopic. +func FanoutOnly() TopicOpt { + return func(t *Topic) error { + t.fanoutOnly = true + return nil + } +} + // WithTopicMessageIdFn sets custom MsgIdFunction for a Topic, enabling topics to have own msg id generation rules. func WithTopicMessageIdFn(msgId MsgIdFunction) TopicOpt { return func(t *Topic) error { diff --git a/topic.go b/topic.go index 7966c651..c41da115 100644 --- a/topic.go +++ b/topic.go @@ -16,6 +16,9 @@ import ( // ErrTopicClosed is returned if a Topic is utilized after it has been closed var ErrTopicClosed = errors.New("this Topic is closed, try opening a new one") +// ErrFanoutOnlyTopic is returned if a relay is requested on a fanout-only topic +var ErrFanoutOnlyTopic = errors.New("cannot relay on a fanout-only topic") + // ErrNilSignKey is returned if a nil private key was provided var ErrNilSignKey = errors.New("nil sign key") @@ -33,6 +36,8 @@ type Topic struct { mux sync.RWMutex closed bool + fanoutOnly bool + requestPartialMessages bool supportsPartialMessages bool } @@ -192,6 +197,9 @@ func (t *Topic) Relay() (RelayCancelFunc, error) { if t.closed { return nil, ErrTopicClosed } + if t.fanoutOnly { + return nil, ErrFanoutOnlyTopic + } out := make(chan RelayCancelFunc, 1)