full close timeout handle (#321)

pkg/hive/hive.go

@@ -118,7 +118,7 @@ func (s *Service) sendPeers(ctx context.Context, peer swarm.Address, peers []swa
 	return stream.FullClose()
 }
 
-func (s *Service) peersHandler(_ context.Context, peer p2p.Peer, stream p2p.Stream) error {
+func (s *Service) peersHandler(ctx context.Context, peer p2p.Peer, stream p2p.Stream) error {
 	_, r := protobuf.NewWriterAndReader(stream)
 	var peersReq pb.Peers
 	if err := r.ReadMsgWithTimeout(messageTimeout, &peersReq); err != nil {
@@ -143,7 +143,7 @@ func (s *Service) peersHandler(_ context.Context, peer p2p.Peer, stream p2p.Stre
 		}
 
 		if s.peerHandler != nil {
-			if err := s.peerHandler(context.Background(), bzzAddress.Overlay); err != nil {
+			if err := s.peerHandler(ctx, bzzAddress.Overlay); err != nil {
 				return err
 			}
 		}

pkg/kademlia/kademlia.go

@@ -297,7 +297,7 @@ func (k *Kad) announce(ctx context.Context, peer swarm.Address) error {
 			return false, false, nil
 		}
 		addrs = append(addrs, connectedPeer)
-		if err := k.discovery.BroadcastPeers(context.Background(), connectedPeer, peer); err != nil {
+		if err := k.discovery.BroadcastPeers(ctx, connectedPeer, peer); err != nil {
 			// we don't want to fail the whole process because of this, keep on gossiping
 			k.logger.Debugf("error gossiping peer %s to peer %s: %v", peer, connectedPeer, err)
 			return false, false, nil
@@ -309,7 +309,7 @@ func (k *Kad) announce(ctx context.Context, peer swarm.Address) error {
 		return nil
 	}
 
-	err := k.discovery.BroadcastPeers(context.Background(), peer, addrs...)
+	err := k.discovery.BroadcastPeers(ctx, peer, addrs...)
 	if err != nil {
 		_ = k.p2p.Disconnect(peer)
 	}

pkg/p2p/libp2p/libp2p.go

@@ -220,11 +220,19 @@ func New(ctx context.Context, signer beecrypto.Signer, networkID uint64, overlay
 		}
 
 		if exists := s.peers.addIfNotExists(stream.Conn(), i.BzzAddress.Overlay); exists {
-			_ = stream.Close()
+			if err = stream.Close(); err != nil {
+				s.logger.Debugf("handshake: could not close stream %s: %v", peerID, err)
+				s.logger.Errorf("unable to handshake with peer %v", peerID)
+				_ = s.disconnect(peerID)
+			}
 			return
 		}
 
-		_ = stream.Close()
+		if err = stream.Close(); err != nil {
+			s.logger.Debugf("handshake: could not close stream %s: %v", peerID, err)
+			s.logger.Errorf("unable to handshake with peer %v", peerID)
+			_ = s.disconnect(peerID)
+		}
 
 		err = s.addressbook.Put(i.BzzAddress.Overlay, *i.BzzAddress)
 		if err != nil {
@@ -366,6 +374,7 @@ func (s *Service) Connect(ctx context.Context, addr ma.Multiaddr) (address *bzz.
 
 	if exists := s.peers.addIfNotExists(stream.Conn(), i.BzzAddress.Overlay); exists {
 		if err := helpers.FullClose(stream); err != nil {
+			_ = s.disconnect(info.ID)
 			return nil, err
 		}
 
@@ -373,6 +382,7 @@ func (s *Service) Connect(ctx context.Context, addr ma.Multiaddr) (address *bzz.
 	}
 
 	if err := helpers.FullClose(stream); err != nil {
+		_ = s.disconnect(info.ID)
 		return nil, err
 	}
 

pkg/topology/full/full.go

@@ -100,7 +100,7 @@ func (d *driver) AddPeer(ctx context.Context, addr swarm.Address) error {
 		}
 
 		connectedAddrs = append(connectedAddrs, addressee.Address)
-		if err := d.discovery.BroadcastPeers(context.Background(), addressee.Address, addr); err != nil {
+		if err := d.discovery.BroadcastPeers(ctx, addressee.Address, addr); err != nil {
 			return err
 		}
 	}
@@ -109,7 +109,7 @@ func (d *driver) AddPeer(ctx context.Context, addr swarm.Address) error {
 		return nil
 	}
 
-	return d.discovery.BroadcastPeers(context.Background(), addr, connectedAddrs...)
+	return d.discovery.BroadcastPeers(ctx, addr, connectedAddrs...)
 }
 
 // ClosestPeer returns the closest connected peer we have in relation to a