added flood publishing

libp2p/pubsub/gossipsub.py

@@ -94,6 +94,8 @@ class GossipSub(IPubsubRouter, Service):
     prune_back_off: int
     unsubscribe_back_off: int
 
+    flood_publish: bool
+
     def __init__(
         self,
         protocols: Sequence[TProtocol],
@@ -112,6 +114,7 @@ def __init__(
         px_peers_count: int = 16,
         prune_back_off: int = 60,
         unsubscribe_back_off: int = 10,
+        flood_publish: bool = False,
     ) -> None:
         self.protocols = list(protocols)
         self.pubsub = None
@@ -152,6 +155,8 @@ def __init__(
         self.prune_back_off = prune_back_off
         self.unsubscribe_back_off = unsubscribe_back_off
 
+        self.flood_publish = flood_publish
+
     async def run(self) -> None:
         self.manager.run_daemon_task(self.heartbeat)
         if len(self.direct_peers) > 0:
@@ -285,42 +290,50 @@ def _get_peers_to_send(
             if topic not in self.pubsub.peer_topics:
                 continue
 
-            # direct peers
-            _direct_peers: set[ID] = {_peer for _peer in self.direct_peers}
-            send_to.update(_direct_peers)
-
-            # floodsub peers
-            floodsub_peers: set[ID] = {
-                peer_id
-                for peer_id in self.pubsub.peer_topics[topic]
-                if self.peer_protocol[peer_id] == floodsub.PROTOCOL_ID
-            }
-            send_to.update(floodsub_peers)
-
-            # gossipsub peers
-            gossipsub_peers: set[ID] = set()
-            if topic in self.mesh:
-                gossipsub_peers = self.mesh[topic]
+            if self.flood_publish and msg_forwarder == self.pubsub.my_id:
+                for peer in self.pubsub.peer_topics[topic]:
+                    # TODO: add score threshold check when peer scoring is implemented
+                    #       if direct peer then skip score check
+                    send_to.add(peer)
             else:
-                # When we publish to a topic that we have not subscribe to, we randomly
-                # pick `self.degree` number of peers who have subscribed to the topic
-                # and add them as our `fanout` peers.
-                topic_in_fanout: bool = topic in self.fanout
-                fanout_peers: set[ID] = self.fanout[topic] if topic_in_fanout else set()
-                fanout_size = len(fanout_peers)
-                if not topic_in_fanout or (
-                    topic_in_fanout and fanout_size < self.degree
-                ):
-                    if topic in self.pubsub.peer_topics:
-                        # Combine fanout peers with selected peers
-                        fanout_peers.update(
-                            self._get_in_topic_gossipsub_peers_from_minus(
-                                topic, self.degree - fanout_size, fanout_peers
+                # direct peers
+                direct_peers: set[ID] = {_peer for _peer in self.direct_peers}
+                send_to.update(direct_peers)
+
+                # floodsub peers
+                floodsub_peers: set[ID] = {
+                    peer_id
+                    for peer_id in self.pubsub.peer_topics[topic]
+                    if self.peer_protocol[peer_id] == floodsub.PROTOCOL_ID
+                }
+                send_to.update(floodsub_peers)
+
+                # gossipsub peers
+                gossipsub_peers: set[ID] = set()
+                if topic in self.mesh:
+                    gossipsub_peers = self.mesh[topic]
+                else:
+                    # When we publish to a topic that we have not subscribe to, we
+                    # randomly pick `self.degree` number of peers who have subscribed
+                    #  to the topic and add them as our `fanout` peers.
+                    topic_in_fanout: bool = topic in self.fanout
+                    fanout_peers: set[ID] = (
+                        self.fanout[topic] if topic_in_fanout else set()
+                    )
+                    fanout_size = len(fanout_peers)
+                    if not topic_in_fanout or (
+                        topic_in_fanout and fanout_size < self.degree
+                    ):
+                        if topic in self.pubsub.peer_topics:
+                            # Combine fanout peers with selected peers
+                            fanout_peers.update(
+                                self._get_in_topic_gossipsub_peers_from_minus(
+                                    topic, self.degree - fanout_size, fanout_peers
+                                )
                             )
-                        )
-                self.fanout[topic] = fanout_peers
-                gossipsub_peers = fanout_peers
-            send_to.update(gossipsub_peers)
+                    self.fanout[topic] = fanout_peers
+                    gossipsub_peers = fanout_peers
+                send_to.update(gossipsub_peers)
         # Excludes `msg_forwarder` and `origin`
         yield from send_to.difference([msg_forwarder, origin])
 

libp2p/tools/constants.py

@@ -45,6 +45,7 @@ class GossipsubParams(NamedTuple):
     px_peers_count: int = 16
     prune_back_off: int = 60
     unsubscribe_back_off: int = 10
+    flood_publish: bool = False
 
 
 GOSSIPSUB_PARAMS = GossipsubParams()

newsfragments/713.feature.rst

@@ -0,0 +1 @@
+Added flood publishing.

tests/core/pubsub/test_gossipsub.py

@@ -594,6 +594,48 @@ async def test_sparse_connect():
 
 
 @pytest.mark.trio
+async def test_flood_publish():
+    async with PubsubFactory.create_batch_with_gossipsub(
+        6,
+        degree=2,
+        degree_low=1,
+        degree_high=3,
+        flood_publish=False,
+    ) as pubsubs_gsub:
+        routers: list[GossipSub] = []
+        for pubsub in pubsubs_gsub:
+            assert isinstance(pubsub.router, GossipSub)
+            routers.append(pubsub.router)
+        hosts = [ps.host for ps in pubsubs_gsub]
+
+        topic = "flood_test_topic"
+        queues = [await pubsub.subscribe(topic) for pubsub in pubsubs_gsub]
+
+        # connect host 0 to all other hosts
+        await one_to_all_connect(hosts, 0)
+
+        # wait for connections to be established
+        await trio.sleep(1)
+
+        # publish a message from the first host
+        msg_content = b"flood_msg"
+        await pubsubs_gsub[0].publish(topic, msg_content)
+
+        # wait for messages to propagate
+        await trio.sleep(0.5)
+
+        print(routers[0].mesh[topic])
+        if routers[0].pubsub:
+            print(routers[0].pubsub.peer_topics)
+
+        # verify all nodes received the message
+        for queue in queues:
+            msg = await queue.get()
+            assert msg.data == msg_content, (
+                f"node did not receive expected message: {msg.data}"
+            )
+
+
 async def test_connect_some_with_fewer_hosts_than_degree():
     """Test connect_some when there are fewer hosts than degree."""
     # Create 3 hosts with degree=5

tests/utils/factories.py

@@ -576,6 +576,7 @@ async def create_batch_with_gossipsub(
         px_peers_count: int = GOSSIPSUB_PARAMS.px_peers_count,
         prune_back_off: int = GOSSIPSUB_PARAMS.prune_back_off,
         unsubscribe_back_off: int = GOSSIPSUB_PARAMS.unsubscribe_back_off,
+        flood_publish: bool = GOSSIPSUB_PARAMS.flood_publish,
         security_protocol: TProtocol | None = None,
         muxer_opt: TMuxerOptions | None = None,
         msg_id_constructor: None
@@ -600,6 +601,7 @@ async def create_batch_with_gossipsub(
                 px_peers_count=px_peers_count,
                 prune_back_off=prune_back_off,
                 unsubscribe_back_off=unsubscribe_back_off,
+                flood_publish=flood_publish,
             )
         else:
             gossipsubs = GossipsubFactory.create_batch(
@@ -618,6 +620,7 @@ async def create_batch_with_gossipsub(
                 px_peers_count=px_peers_count,
                 prune_back_off=prune_back_off,
                 unsubscribe_back_off=unsubscribe_back_off,
+                flood_publish=flood_publish,
             )
 
         async with cls._create_batch_with_router(