rename {Add/Remove}Peer to On{New/Closed}OutboundStream

Makes the name reflect the thing that this lifecycle is attached to.

AddPeer/RemovePeer makes it sound as if the lifecycle is attached to the
underlying peer connection. That's not the case. The lifecycle here is
only attached to the outbound stream.

Author: MarcoPolo

extensions.go

@@ -76,7 +76,7 @@ func (pe *PeerExtensions) ExtendRPC(rpc *RPC) *RPC {
 // Purposely not trying to make a generic extension interface as there is only
 // one real consumer (partial messages). This may change in the future.
 type partialMessageInterface interface {
-	RemovePeer(peer.ID)
+	OnClosedOutboundStream(peer.ID)
 	HandleRPC(from peer.ID, rpc *pubsub_pb.PartialMessagesExtension) error
 	Heartbeat()
 	EmitGossip(topic string, peers []peer.ID)
@@ -112,7 +112,7 @@ func (es *extensionsState) HandleRPC(rpc *RPC) error {
 		if _, ok := es.sentExtensions[rpc.from]; ok {
 			// We just finished both sending and receiving the extensions
 			// control message.
-			es.extensionsAddPeer(rpc.from)
+			es.extensionsOnNewOutboundStream(rpc.from)
 		}
 	} else {
 		// We already have an extension for this peer. If they send us another
@@ -136,44 +136,44 @@ func (es *extensionsState) OnClosedIncomingStream(id peer.ID, _ protocol.ID) {
 	}
 }
 
-func (es *extensionsState) AddPeer(id peer.ID, helloPacket *RPC) *RPC {
+func (es *extensionsState) OnNewOutboundStream(id peer.ID, helloPacket *RPC) *RPC {
 	// Send our extensions as the first message.
 	helloPacket = es.myExtensions.ExtendRPC(helloPacket)
 
 	es.sentExtensions[id] = struct{}{}
 	if _, ok := es.peerExtensions[id]; ok {
 		// We've just finished sending and receiving the extensions control
 		// message.
-		es.extensionsAddPeer(id)
+		es.extensionsOnNewOutboundStream(id)
 	}
 	return helloPacket
 }
 
-func (es *extensionsState) RemovePeer(id peer.ID) {
+func (es *extensionsState) OnClosedOutboundStream(id peer.ID) {
 	_, recvdExt := es.peerExtensions[id]
 	_, sentExt := es.sentExtensions[id]
 	if recvdExt && sentExt {
 		// Add peer was previously called, so we need to call remove peer
-		es.extensionsRemovePeer(id)
+		es.extensionsOnClosedOutboundStream(id)
 	}
 	delete(es.sentExtensions, id)
 	if len(es.sentExtensions) == 0 {
 		es.sentExtensions = make(map[peer.ID]struct{})
 	}
 }
 
-// extensionsAddPeer is only called once we've both sent and received the
+// extensionsOnNewOutboundStream is only called once we've both sent and received the
 // extensions control message.
-func (es *extensionsState) extensionsAddPeer(id peer.ID) {
+func (es *extensionsState) extensionsOnNewOutboundStream(id peer.ID) {
 	if es.myExtensions.TestExtension && es.peerExtensions[id].TestExtension {
-		es.testExtension.AddPeer(id)
+		es.testExtension.OnNewOutboundStream(id)
 	}
 }
 
-// extensionsRemovePeer is always called after extensionsAddPeer.
-func (es *extensionsState) extensionsRemovePeer(id peer.ID) {
+// extensionsOnClosedOutboundStream is always called after extensionsOnNewOutboundStream.
+func (es *extensionsState) extensionsOnClosedOutboundStream(id peer.ID) {
 	if es.myExtensions.PartialMessages && es.peerExtensions[id].PartialMessages {
-		es.partialMessagesExtension.RemovePeer(id)
+		es.partialMessagesExtension.OnClosedOutboundStream(id)
 	}
 }
 

floodsub.go

@@ -44,13 +44,13 @@ func (fs *FloodSubRouter) Attach(p *PubSub) {
 func (fs *FloodSubRouter) OnNewIncomingStream(peer.ID, protocol.ID)    {}
 func (fs *FloodSubRouter) OnClosedIncomingStream(peer.ID, protocol.ID) {}
 
-func (fs *FloodSubRouter) AddPeer(p peer.ID, proto protocol.ID, hello *RPC) *RPC {
-	fs.tracer.AddPeer(p, proto)
+func (fs *FloodSubRouter) OnNewOutboundStream(p peer.ID, proto protocol.ID, hello *RPC) *RPC {
+	fs.tracer.OnNewOutboundStream(p, proto)
 	return hello
 }
 
-func (fs *FloodSubRouter) RemovePeer(p peer.ID) {
-	fs.tracer.RemovePeer(p)
+func (fs *FloodSubRouter) OnClosedOutboundStream(p peer.ID) {
+	fs.tracer.OnClosedOutboundStream(p)
 }
 
 func (fs *FloodSubRouter) EnoughPeers(topic string, suggested int) bool {

gossip_tracer.go

@@ -167,17 +167,17 @@ func (gt *gossipTracer) ValidateMessage(msg *Message) {
 	gt.fulfillPromise(msg)
 }
 
-func (gt *gossipTracer) AddPeer(p peer.ID, proto protocol.ID) {}
-func (gt *gossipTracer) RemovePeer(p peer.ID)                 {}
-func (gt *gossipTracer) Join(topic string)                    {}
-func (gt *gossipTracer) Leave(topic string)                   {}
-func (gt *gossipTracer) Graft(p peer.ID, topic string)        {}
-func (gt *gossipTracer) Prune(p peer.ID, topic string)        {}
-func (gt *gossipTracer) DuplicateMessage(msg *Message)        {}
-func (gt *gossipTracer) RecvRPC(rpc *RPC)                     {}
-func (gt *gossipTracer) SendRPC(rpc *RPC, p peer.ID)          {}
-func (gt *gossipTracer) DropRPC(rpc *RPC, p peer.ID)          {}
-func (gt *gossipTracer) UndeliverableMessage(msg *Message)    {}
+func (gt *gossipTracer) OnNewOutboundStream(p peer.ID, proto protocol.ID) {}
+func (gt *gossipTracer) OnClosedOutboundStream(p peer.ID)                 {}
+func (gt *gossipTracer) Join(topic string)                                {}
+func (gt *gossipTracer) Leave(topic string)                               {}
+func (gt *gossipTracer) Graft(p peer.ID, topic string)                    {}
+func (gt *gossipTracer) Prune(p peer.ID, topic string)                    {}
+func (gt *gossipTracer) DuplicateMessage(msg *Message)                    {}
+func (gt *gossipTracer) RecvRPC(rpc *RPC)                                 {}
+func (gt *gossipTracer) SendRPC(rpc *RPC, p peer.ID)                      {}
+func (gt *gossipTracer) DropRPC(rpc *RPC, p peer.ID)                      {}
+func (gt *gossipTracer) UndeliverableMessage(msg *Message)                {}
 
 func (gt *gossipTracer) ThrottlePeer(p peer.ID) {
 	gt.Lock()

gossipsub.go

@@ -758,9 +758,9 @@ func (gs *GossipSubRouter) OnClosedIncomingStream(pid peer.ID, proto protocol.ID
 	}
 }
 
-func (gs *GossipSubRouter) AddPeer(p peer.ID, proto protocol.ID, helloPacket *RPC) *RPC {
+func (gs *GossipSubRouter) OnNewOutboundStream(p peer.ID, proto protocol.ID, helloPacket *RPC) *RPC {
 	gs.logger.Debug("PEERUP: Add new peer using protocol", "peer", p, "protocol", proto)
-	gs.tracer.AddPeer(p, proto)
+	gs.tracer.OnNewOutboundStream(p, proto)
 	gs.peers[p] = proto
 
 	// track the connection direction
@@ -786,16 +786,16 @@ loop:
 	}
 	gs.outbound[p] = outbound
 	if gs.feature(GossipSubFeatureExtensions, proto) {
-		helloPacket = gs.extensions.AddPeer(p, helloPacket)
+		helloPacket = gs.extensions.OnNewOutboundStream(p, helloPacket)
 	}
 	return helloPacket
 }
 
-func (gs *GossipSubRouter) RemovePeer(p peer.ID) {
+func (gs *GossipSubRouter) OnClosedOutboundStream(p peer.ID) {
 	gs.logger.Debug("PEERDOWN: Remove disconnected peer", "peer", p)
-	gs.tracer.RemovePeer(p)
+	gs.tracer.OnClosedOutboundStream(p)
 	if gs.feature(GossipSubFeatureExtensions, gs.peers[p]) {
-		gs.extensions.RemovePeer(p)
+		gs.extensions.OnClosedOutboundStream(p)
 	}
 	delete(gs.peers, p)
 	for _, peers := range gs.mesh {

gossipsub_test.go

@@ -3102,20 +3102,20 @@ func (m *mockRawTracer) RecvRPC(rpc *RPC) {
 	}
 }
 
-func (m *mockRawTracer) AddPeer(p peer.ID, proto protocol.ID)      {}
-func (m *mockRawTracer) DeliverMessage(msg *Message)               {}
-func (m *mockRawTracer) DropRPC(rpc *RPC, p peer.ID)               {}
-func (m *mockRawTracer) DuplicateMessage(msg *Message)             {}
-func (m *mockRawTracer) Graft(p peer.ID, topic string)             {}
-func (m *mockRawTracer) Join(topic string)                         {}
-func (m *mockRawTracer) Leave(topic string)                        {}
-func (m *mockRawTracer) Prune(p peer.ID, topic string)             {}
-func (m *mockRawTracer) RejectMessage(msg *Message, reason string) {}
-func (m *mockRawTracer) RemovePeer(p peer.ID)                      {}
-func (m *mockRawTracer) SendRPC(rpc *RPC, p peer.ID)               {}
-func (m *mockRawTracer) ThrottlePeer(p peer.ID)                    {}
-func (m *mockRawTracer) UndeliverableMessage(msg *Message)         {}
-func (m *mockRawTracer) ValidateMessage(msg *Message)              {}
+func (m *mockRawTracer) OnNewOutboundStream(p peer.ID, proto protocol.ID) {}
+func (m *mockRawTracer) DeliverMessage(msg *Message)                      {}
+func (m *mockRawTracer) DropRPC(rpc *RPC, p peer.ID)                      {}
+func (m *mockRawTracer) DuplicateMessage(msg *Message)                    {}
+func (m *mockRawTracer) Graft(p peer.ID, topic string)                    {}
+func (m *mockRawTracer) Join(topic string)                                {}
+func (m *mockRawTracer) Leave(topic string)                               {}
+func (m *mockRawTracer) Prune(p peer.ID, topic string)                    {}
+func (m *mockRawTracer) RejectMessage(msg *Message, reason string)        {}
+func (m *mockRawTracer) OnClosedOutboundStream(p peer.ID)                 {}
+func (m *mockRawTracer) SendRPC(rpc *RPC, p peer.ID)                      {}
+func (m *mockRawTracer) ThrottlePeer(p peer.ID)                           {}
+func (m *mockRawTracer) UndeliverableMessage(msg *Message)                {}
+func (m *mockRawTracer) ValidateMessage(msg *Message)                     {}
 
 var _ RawTracer = &mockRawTracer{}
 

partialmessages/partialmsgs.go

@@ -241,13 +241,13 @@ func (e *PartialMessagesExtension[PeerState]) initPeerState(topic string, gState
 	}
 }
 
-func (e *PartialMessagesExtension[PeerState]) RemovePeer(id peer.ID) {
+func (e *PartialMessagesExtension[PeerState]) OnClosedOutboundStream(id peer.ID) {
 	for topic, tState := range e.statePerTopicPerGroup {
 		for _, gState := range tState {
 			delete(gState.peerState, id)
 		}
 		if ctr, ok := e.peerInitiatedGroupCounter[topic]; ok {
-			ctr.RemovePeer(id)
+			ctr.OnClosedOutboundStream(id)
 		}
 	}
 }
@@ -357,7 +357,7 @@ func (ctr *peerInitiatedGroupCounterState) Dec(id peer.ID) {
 	}
 }
 
-func (ctr *peerInitiatedGroupCounterState) RemovePeer(id peer.ID) {
+func (ctr *peerInitiatedGroupCounterState) OnClosedOutboundStream(id peer.ID) {
 	if n, ok := ctr.perPeer[id]; ok {
 		ctr.total -= n
 		delete(ctr.perPeer, id)

partialmessages/partialmsgs_test.go

@@ -70,8 +70,8 @@ func (m *mockNetworkPartialMessages) removePeers() {
 			if a == b {
 				continue
 			}
-			m.handlers[a].RemovePeer(b)
-			m.handlers[b].RemovePeer(a)
+			m.handlers[a].OnClosedOutboundStream(b)
+			m.handlers[b].OnClosedOutboundStream(a)
 		}
 	}
 
@@ -1168,8 +1168,8 @@ func TestGossipDelivery(t *testing.T) {
 	}
 
 	// Cleanup
-	h1Handler.RemovePeer(h2ID)
-	h2Handler.RemovePeer(h1ID)
+	h1Handler.OnClosedOutboundStream(h2ID)
+	h2Handler.OnClosedOutboundStream(h1ID)
 	for range 10 {
 		h1Handler.Heartbeat()
 		h2Handler.Heartbeat()
@@ -1273,10 +1273,10 @@ func TestPeerInitiatedCounter(t *testing.T) {
 	}
 
 	// All peers go away, and the counts should be back to 0
-	handler.RemovePeer("1")
+	handler.OnClosedOutboundStream("1")
 	for id := range 2 {
 		otherPeer := fmt.Sprintf("peer%d", id)
-		handler.RemovePeer(peer.ID(otherPeer))
+		handler.OnClosedOutboundStream(peer.ID(otherPeer))
 	}
 
 	assertCounts(0, map[peer.ID]int{})
@@ -1376,7 +1376,7 @@ func FuzzPeerInitiatedCounter(f *testing.F) {
 					delete(expectedPeercounts, peer.ID(otherPeer))
 				}
 
-				handler.RemovePeer(peer.ID(otherPeer))
+				handler.OnClosedOutboundStream(peer.ID(otherPeer))
 			case 2: // heartbeat until everything is cleared
 				script = script[1:]
 				for range handler.GroupTTLByHeatbeat + 1 {