fix: modify max packet size

go.mod

@@ -127,7 +127,7 @@ require (
 	github.com/mmcloughlin/addchain v0.4.0 // indirect
 	github.com/naoina/go-stringutil v0.1.0 // indirect
 	github.com/opentracing/opentracing-go v1.1.0 // indirect
-	github.com/optimism-java/utp-go v0.0.0-20231111152515-b2c1e9aba225 // indirect
+	github.com/optimism-java/utp-go v0.0.0-20231114114639-92925ba7e35e // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.12.0 // indirect

go.sum

@@ -497,6 +497,8 @@ github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98 h1:uxUbd8LFc2
 github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
 github.com/optimism-java/utp-go v0.0.0-20231111152515-b2c1e9aba225 h1:UUVmsVAv/4v0TMW3AFPwxAkuyNjKsAWVyqxfR10gMGE=
 github.com/optimism-java/utp-go v0.0.0-20231111152515-b2c1e9aba225/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
+github.com/optimism-java/utp-go v0.0.0-20231114114639-92925ba7e35e h1:61Mw2nE4trMg/Ze/0oFD5o5yKog2UuLUlbUnBS0lCh8=
+github.com/optimism-java/utp-go v0.0.0-20231114114639-92925ba7e35e/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM=
 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0=

p2p/discover/portal_protocol.go

@@ -7,6 +7,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"go.uber.org/zap"
 	"io"
 	"net"
 	"sort"
@@ -165,7 +166,6 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) {
 
 	p.packetRouter = utp.NewSocketRouter(
 		func(buf []byte, addr *net.UDPAddr) (int, error) {
-
 			nodes := p.table.Nodes()
 			var target *enode.Node
 			for _, node := range nodes {
@@ -174,24 +174,28 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) {
 				}
 				if addr.IP != nil && addr.IP.To4().String() == node.IP().To4().String() {
 					target = node
-					p.log.Trace("target info", "ip", node.IP().To4().String(), "port", node.UDP(), "bufLength", len(buf))
+
 					break
 				}
 				if addr.IP == nil {
 					nodeIp := node.IP().To4().String()
 					if nodeIp == "127.0.0.1" || nodeIp == "0.0.0.0" {
 						target = node
-						p.log.Trace("target info", "ip", nodeIp, "port", node.UDP(), "bufLength", len(buf))
 						break
 					}
 				}
 			}
 
+			p.log.Trace("send to target data", "ip", target.IP().String(), "port", target.UDP(), "bufLength", len(buf))
 			_, err := p.DiscV5.TalkRequest(target, portalwire.UTPNetwork, buf)
 			return len(buf), err
 		})
 
-	p.utpSm, err = utp.NewSocketManager("utp", laddr, utp.WithPacketRouter(p.packetRouter), utp.WithBlockPacketCount(50))
+	logger, err := zap.NewDevelopmentConfig().Build()
+	if err != nil {
+		return nil, err
+	}
+	p.utpSm, err = utp.NewSocketManager("utp", laddr, utp.WithLogger(logger.Named(listenAddr)), utp.WithPacketRouter(p.packetRouter), utp.WithBlockPacketCount(50), utp.WithMaxPacketSize(1145))
 	if err != nil {
 		return nil, err
 	}
@@ -350,7 +354,7 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
 		laddr := p.utp.Addr().(*utp.Addr)
 		raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()}
 		connId := binary.BigEndian.Uint16(connIdMsg.Id[:])
-		conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(rctx), utp.WithConnId(uint32(connId)))
+		conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(rctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId)))
 		if err != nil {
 			rcancel()
 			return 0xff, nil, err
@@ -363,8 +367,8 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
 		}
 		// Read ALL the data from the connection until EOF and return it
 		data := make([]byte, 0)
+		buf := make([]byte, 1024)
 		for {
-			buf := make([]byte, 1024)
 			var n int
 			n, err = conn.Read(buf)
 			if err != nil {
@@ -473,9 +477,7 @@ func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, ms
 	if n := p.DiscV5.getNode(id); n != nil {
 		p.table.addSeenNode(wrapNode(n))
 	}
-	if len(msg) == 0 {
+	p.log.Trace("receive utp data", "addr", addr, "msg-length", len(msg))
-		fmt.Println("receive a emtpy msg")
-	}
 	p.packetRouter.ReceiveMessage(msg, addr)
 	return []byte("")
 }
@@ -695,8 +697,14 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
 			ctx, cancel := context.WithTimeout(context.Background(), defaultUTPConnectTimeout)
 			var conn *utp.Conn
 			conn, err = p.utp.AcceptUTPContext(ctx, connIdSend)
+			defer func(conn *utp.Conn) {
+				err := conn.Close()
+				if err != nil {
+					p.log.Error("failed to close utp connection", "err", err)
+				}
+			}(conn)
 			if err != nil {
-				p.log.Error("failed to accept utp connection", "err", err)
+				p.log.Error("failed to accept utp connection", "connId", connIdSend, "err", err)
 				cancel()
 				return
 			}

p2p/discover/portal_protocol_test.go

@@ -2,8 +2,11 @@ package discover
 
 import (
 	"crypto/rand"
+	"errors"
 	"fmt"
 	"github.com/optimism-java/utp-go"
+	"io"
+	"net"
 	"sync"
 	"testing"
 	"time"
@@ -88,7 +91,10 @@ func TestPortalWireProtocolUdp(t *testing.T) {
 	cliSendMsgWithRandomCid := "there are connection id: random!"
 
 	serverEchoWithCid := "accept connection sends back msg: echo"
-	serverEchoWithRandomCid := "ccept connection with random cid sends msg: echo"
+	//serverEchoWithRandomCid := "ccept connection with random cid sends msg: echo"
+
+	largeTestContent := make([]byte, 1199)
+	_, err = rand.Read(largeTestContent)
 	go func() {
 		var acceptConn *utp.Conn
 		defer func() {
@@ -108,7 +114,11 @@ func TestPortalWireProtocolUdp(t *testing.T) {
 		acceptConn.Write([]byte(serverEchoWithCid))
 	}()
 	go func() {
-		defer wg.Done()
+		var randomConnIdConn net.Conn
+		defer func() {
+			wg.Done()
+			_ = randomConnIdConn.Close()
+		}()
 		randomConnIdConn, err := node1.utp.Accept()
 		if err != nil {
 			panic(err)
@@ -119,11 +129,16 @@ func TestPortalWireProtocolUdp(t *testing.T) {
 			panic(err)
 		}
 		assert.Equal(t, cliSendMsgWithRandomCid, string(buf[:n]))
-		randomConnIdConn.Write([]byte(serverEchoWithRandomCid))
+
+		randomConnIdConn.Write(largeTestContent)
 	}()
 
 	go func() {
-		defer wg.Done()
+		var connWithConnId net.Conn
+		defer func() {
+			wg.Done()
+			//_ = connWithConnId.Close()
+		}()
 		connWithConnId, err := utp.DialUTPOptions("utp", lAddr, rAddr, utp.WithConnId(cid), utp.WithSocketManager(node2.utpSm))
 		if err != nil {
 			panic(err)
@@ -140,7 +155,11 @@ func TestPortalWireProtocolUdp(t *testing.T) {
 		assert.Equal(t, serverEchoWithCid, string(buf[:n]))
 	}()
 	go func() {
-		defer wg.Done()
+		var randomConnIdConn net.Conn
+		defer func() {
+			wg.Done()
+			//_ = randomConnIdConn.Close()
+		}()
 		randomConnIdConn, err := utp.DialUTPOptions("utp", lAddr, rAddr, utp.WithSocketManager(node2.utpSm))
 		if err != nil {
 			panic(err)
@@ -149,12 +168,20 @@ func TestPortalWireProtocolUdp(t *testing.T) {
 		if err != nil {
 			panic(err)
 		}
-		buf := make([]byte, 100)
+
-		n, err := randomConnIdConn.Read(buf)
+		data := make([]byte, 0)
-		if err != nil {
+		buf := make([]byte, 1024)
-			panic(err)
+		for {
+			var n int
+			n, err = randomConnIdConn.Read(buf)
+			if err != nil {
+				if errors.Is(err, io.EOF) {
+					break
+				}
+			}
+			data = append(data, buf[:n]...)
 		}
-		assert.Equal(t, serverEchoWithRandomCid, string(buf[:n]))
+		assert.Equal(t, largeTestContent, data)
 	}()
 	wg.Wait()
 }
@@ -232,6 +259,6 @@ func TestPortalWireProtocol(t *testing.T) {
 
 	flag, content, err = node2.findContent(node1.localNode.Node(), []byte("large_test_key"))
 	assert.NoError(t, err)
-	assert.Equal(t, portalwire.ContentConnIdSelector, flag)
 	assert.Equal(t, largeTestContent, content)
+	assert.Equal(t, portalwire.ContentConnIdSelector, flag)
 }