feat: FanoutOnly topic option

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
 	}
 

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)
+	}
+}

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 {

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)