fix: if we do not send a message to a peer, remove them from the recipients list

Author: achingbrain

test/go-gossipsub.spec.ts

@@ -269,7 +269,6 @@ describe('go-libp2p-pubsub gossipsub tests', function () {
       sendRecv.push(results)
     }
     await Promise.all(sendRecv)
-
   })
 
   it('test gossipsub fanout maintenance', async function () {
@@ -332,7 +331,6 @@ describe('go-libp2p-pubsub gossipsub tests', function () {
     sendRecv = []
     await sendMessages(2)
     await Promise.all(sendRecv)
-
   })
 
   it('test gossipsub fanout expiry', async function () {
@@ -381,8 +379,6 @@ describe('go-libp2p-pubsub gossipsub tests', function () {
     await pWaitFor(async () => {
       return (psubs[0].getPubSub() as GossipSub).fanout.size === 0
     })
-
-
   })
 
   it('test gossipsub gossip', async function () {
@@ -421,7 +417,6 @@ describe('go-libp2p-pubsub gossipsub tests', function () {
     }
     // and wait for some gossip flushing
     await Promise.all(psubs.map(async (ps) => await awaitEvents(ps.getPubSub(), 'gossipsub:heartbeat', 2)))
-
   })
 
   it('test gossipsub gossip propagation', async function () {

test/gossip.spec.ts

@@ -111,6 +111,9 @@ describe('gossip', () => {
     // should have peerB as a subscriber to the topic
     expect(nodeA.getPubSub().getSubscribers(topic).map(p => p.toString())).to.include(peerB, 'did not know about peerB\'s subscription to topic')
 
+    // should be able to send them messages
+    expect((nodeA.getPubSub() as GossipSub).peers.get(peerB)).to.have.property('isWritable', true, 'nodeA did not have connection open to peerB')
+
     // set spy. NOTE: Forcing private property to be public
     const nodeASpy = sinon.spy(nodeA.getPubSub() as GossipSub, 'piggybackControl')
 
@@ -121,7 +124,7 @@ describe('gossip', () => {
     const publishResult = await nodeA.getPubSub().publish(topic, uint8ArrayFromString('hey'))
 
     // should have sent message to peerB
-    expect(publishResult.recipients.map(p => p.toString())).to.include(peerB, 'did not sent pubsub message to peerB')
+    expect(publishResult.recipients.map(p => p.toString())).to.include(peerB, 'did not send pubsub message to peerB')
 
     expect(nodeASpy.callCount).to.be.equal(1)
     // expect control message to be sent alongside published message

ts/index.ts

@@ -1875,10 +1875,16 @@ export class GossipSub extends EventEmitter<GossipsubEvents> implements Initiali
 
     // Send to set of peers aggregated from direct, mesh, fanout
     const rpc = createGossipRpc([rawMsg])
-    tosend.forEach((id) => {
+
+    for (const id of tosend) {
       // self.send_message(*peer_id, event.clone())?;
-      this.sendRpc(id, rpc)
-    })
+      const sent = this.sendRpc(id, rpc)
+
+      // did not actually send the message
+      if (!sent) {
+        tosend.delete(id)
+      }
+    }
 
     this.metrics?.onPublishMsg(topic, tosendCount, tosend.size, rawMsg.data != null ? rawMsg.data.length : 0)
 
@@ -1990,11 +1996,11 @@ export class GossipSub extends EventEmitter<GossipsubEvents> implements Initiali
   /**
    * Send an rpc object to a peer
    */
-  private sendRpc(id: PeerIdStr, rpc: RPC): void {
+  private sendRpc(id: PeerIdStr, rpc: RPC): boolean {
     const peerStreams = this.peers.get(id)
     if (!peerStreams || !peerStreams.isWritable) {
       this.log(`Cannot send RPC to ${id} as there is no open stream to it available`)
-      return
+      return false
     }
 
     // piggyback control message retries
@@ -2015,6 +2021,8 @@ export class GossipSub extends EventEmitter<GossipsubEvents> implements Initiali
     peerStreams.write(rpcBytes)
 
     this.metrics?.onRpcSent(rpc, rpcBytes.length)
+
+    return true
   }
 
   public piggybackControl(id: PeerIdStr, outRpc: RPC, ctrl: RPC.ControlMessage): void {