feat:add find content

Signed-off-by: grapebaba <281165273@qq.com>

go.mod

@@ -54,6 +54,7 @@ require (
 	github.com/mattn/go-isatty v0.0.17
 	github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416
 	github.com/olekukonko/tablewriter v0.0.5
+	github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98
 	github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7
 	github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7
 	github.com/rs/cors v1.7.0
@@ -71,7 +72,7 @@ require (
 	golang.org/x/sys v0.13.0
 	golang.org/x/text v0.13.0
 	golang.org/x/time v0.3.0
-	golang.org/x/tools v0.13.0
+	golang.org/x/tools v0.14.0
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -127,7 +128,6 @@ 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-20231024092003-1dd76611b5f2 // 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
@@ -140,10 +140,9 @@ require (
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
-	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/multierr v1.6.0 // indirect
+	go.uber.org/zap v1.26.0 // indirect
-	go.uber.org/zap v1.19.0 // indirect
+	golang.org/x/mod v0.13.0 // indirect
-	golang.org/x/mod v0.12.0 // indirect
 	golang.org/x/net v0.17.0 // indirect
 	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

go.sum

@@ -98,8 +98,6 @@ github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
-github.com/bits-and-blooms/bitset v1.7.0 h1:YjAGVd3XmtK9ktAbX8Zg2g2PwLIMjGREZJHlV4j7NEo=
-github.com/bits-and-blooms/bitset v1.7.0/go.mod h1:gIdJ4wp64HaoK2YrL1Q5/N7Y16edYb8uY+O0FJTyyDA=
 github.com/bits-and-blooms/bitset v1.10.0 h1:ePXTeiPEazB5+opbv5fr8umg2R/1NlzgDsyepwsSr88=
 github.com/bits-and-blooms/bitset v1.10.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
 github.com/btcsuite/btcd/btcec/v2 v2.2.0 h1:fzn1qaOt32TuLjFlkzYSsBC35Q3KUjT1SwPxiMSCF5k=
@@ -499,8 +497,8 @@ github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
-github.com/optimism-java/utp-go v0.0.0-20231024092003-1dd76611b5f2 h1:1H9unjvDxqDVTF9rh99sfolMtp8zYCUNAq+aoeGCxLA=
+github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98 h1:uxUbd8LFc24XetNFjTu9Kp9MqF2zKF92UMbcDuPxYZ8=
-github.com/optimism-java/utp-go v0.0.0-20231024092003-1dd76611b5f2/go.mod h1:ohCuwoc66lfiNpo2Wk22zV07IbB0gte8+TYiclv9an4=
+github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98/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,11 +7,13 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"sort"
 	"time"
 
 	"github.com/VictoriaMetrics/fastcache"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover/portalwire"
 	"github.com/ethereum/go-ethereum/p2p/enode"
@@ -43,9 +45,11 @@ const (
 
 	portalFindnodesResultLimit = 32
 
-	defaultUTPAcceptTimeout = 15 * time.Second
+	defaultUTPConnectTimeout = 15 * time.Second
 
 	defaultUTPWriteTimeout = 60 * time.Second
+
+	defaultUTPReadTimeout = 60 * time.Second
 )
 
 type PortalProtocolConfig struct {
@@ -94,7 +98,7 @@ type PortalProtocol struct {
 	storage        Storage
 }
 
-func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateKey *ecdsa.PrivateKey) (*PortalProtocol, error) {
+func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateKey *ecdsa.PrivateKey, storage Storage) (*PortalProtocol, error) {
 	nodeDB, err := enode.OpenDB(config.NodeDBPath)
 	if err != nil {
 		return nil, err
@@ -117,6 +121,7 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateK
 		cancelCloseCtx: cancelCloseCtx,
 		localNode:      localNode,
 		validSchemes:   enode.ValidSchemes,
+		storage:        storage,
 	}
 
 	return protocol, nil
@@ -161,9 +166,8 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) {
 	p.utpPackets = make(chan *utp.UdpMessage, 10)
 	p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithCustomHandler(
 		func(buf []byte, addr *net.UDPAddr) (int, error) {
-			var a [32]byte
+			id := crypto.Keccak256([]byte(addr.String()))
-			// todo need to find enode.ID by addr
+			_, err := p.DiscV5.TalkRequestToID(enode.ID(id), addr, portalwire.UTPNetwork, buf)
-			_, err := p.DiscV5.TalkRequestToID(a, addr, portalwire.UTPNetwork, buf)
 			return 0, err
 		},
 		func() ([]byte, *net.UDPAddr, error) {
@@ -211,6 +215,44 @@ func (p *PortalProtocol) setupDiscV5AndTable() error {
 	return nil
 }
 
+func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) {
+	enrSeq := p.DiscV5.LocalNode().Seq()
+	radiusBytes, err := p.nodeRadius.MarshalSSZ()
+	if err != nil {
+		return 0, err
+	}
+	customPayload := &portalwire.PingPongCustomData{
+		Radius: radiusBytes,
+	}
+
+	customPayloadBytes, err := customPayload.MarshalSSZ()
+	if err != nil {
+		return 0, err
+	}
+
+	pingRequest := &portalwire.Ping{
+		EnrSeq:        enrSeq,
+		CustomPayload: customPayloadBytes,
+	}
+
+	p.log.Trace("Sending ping request", "protocol", p.protocolId, "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest)
+	pingRequestBytes, err := pingRequest.MarshalSSZ()
+	if err != nil {
+		return 0, err
+	}
+
+	talkRequestBytes := make([]byte, 0, len(pingRequestBytes)+1)
+	talkRequestBytes = append(talkRequestBytes, portalwire.PING)
+	talkRequestBytes = append(talkRequestBytes, pingRequestBytes...)
+
+	talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes)
+
+	if err != nil {
+		p.replaceNode(node)
+	}
+	return p.processPong(node, talkResp)
+}
+
 func (p *PortalProtocol) findNodes(node *enode.Node, distances []uint) ([]*enode.Node, error) {
 	distancesBytes := make([][2]byte, len(distances))
 	for i, distance := range distances {
@@ -241,44 +283,148 @@ func (p *PortalProtocol) findNodes(node *enode.Node, distances []uint) ([]*enode
 	return p.processNodes(node, talkResp, distances)
 }
 
-func (p *PortalProtocol) processNodes(target *enode.Node, resp []byte, distances []uint) ([]*enode.Node, error) {
+func (p *PortalProtocol) findContent(node *enode.Node, contentKey []byte) (byte, interface{}, error) {
-	var (
+	findContent := &portalwire.FindContent{
-		nodes    []*enode.Node
+		ContentKey: contentKey,
-		seen     = make(map[enode.ID]struct{})
+	}
-		err      error
-		verified = 0
-	)
 
+	p.log.Trace("Sending find content request", "id", node.ID(), "findContent", findContent)
+	findContentBytes, err := findContent.MarshalSSZ()
+	if err != nil {
+		p.log.Error("failed to marshal find content request", "err", err)
+		return 0xff, nil, err
+	}
+
+	talkRequestBytes := make([]byte, 0, len(findContentBytes)+1)
+	talkRequestBytes = append(talkRequestBytes, portalwire.FINDCONTENT)
+	talkRequestBytes = append(talkRequestBytes, findContentBytes...)
+
+	talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes)
+	if err != nil {
+		p.log.Error("failed to send find content request", "err", err)
+		return 0xff, nil, err
+	}
+
+	return p.processContent(node, talkResp)
+}
+
+func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte, interface{}, error) {
+	if resp[0] != portalwire.CONTENT {
+		return 0xff, nil, fmt.Errorf("invalid content response")
+	}
+
+	switch resp[1] {
+	case portalwire.ContentRawSelector:
+		content := &portalwire.Content{}
+		err := content.UnmarshalSSZ(resp[2:])
+		if err != nil {
+			return 0xff, nil, err
+		}
+
+		p.log.Trace("Received content response", "id", target.ID(), "content", content)
+		return resp[1], content.Content, nil
+	case portalwire.ContentConnIdSelector:
+		connIdMsg := &portalwire.ConnectionId{}
+		err := connIdMsg.UnmarshalSSZ(resp[2:])
+		if err != nil {
+			return 0xff, nil, err
+		}
+
+		p.log.Trace("Received content response", "id", target.ID(), "connIdMsg", connIdMsg)
+		rctx, rcancel := context.WithTimeout(context.Background(), defaultUTPConnectTimeout)
+		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)))
+		if err != nil {
+			rcancel()
+			return 0xff, nil, err
+		}
+
+		err = conn.SetReadDeadline(time.Now().Add(defaultUTPReadTimeout))
+		if err != nil {
+			rcancel()
+			return 0xff, nil, err
+		}
+		// Read ALL the data from the connection until EOF and return it
+		data := make([]byte, 0)
+		for {
+			buf := make([]byte, 1024)
+			var n int
+			n, err = conn.Read(buf)
+			if err != nil {
+				rcancel()
+				if errors.Is(err, io.EOF) {
+					p.log.Trace("Received content response", "id", target.ID(), "data", data, "size", n)
+					return resp[1], data, nil
+				}
+
+				p.log.Error("failed to read from utp connection", "err", err)
+				return 0xff, nil, err
+			}
+			data = append(data, buf[:n]...)
+		}
+	case portalwire.ContentEnrsSelector:
+		enrs := &portalwire.Enrs{}
+		err := enrs.UnmarshalSSZ(resp[2:])
+
+		if err != nil {
+			return 0xff, nil, err
+		}
+
+		p.log.Trace("Received content response", "id", target.ID(), "enrs", enrs)
+
+		nodes := p.filterNodes(target, enrs.Enrs, nil)
+		return resp[1], nodes, nil
+	default:
+		return 0xff, nil, fmt.Errorf("invalid content response")
+	}
+}
+
+func (p *PortalProtocol) processNodes(target *enode.Node, resp []byte, distances []uint) ([]*enode.Node, error) {
 	if resp[0] != portalwire.NODES {
 		return nil, fmt.Errorf("invalid nodes response")
 	}
 
 	nodesResp := &portalwire.Nodes{}
-	err = nodesResp.UnmarshalSSZ(resp[1:])
+	err := nodesResp.UnmarshalSSZ(resp[1:])
 	if err != nil {
 		return nil, err
 	}
 
 	p.table.addVerifiedNode(wrapNode(target))
-	var n *enode.Node
+	nodes := p.filterNodes(target, nodesResp.Enrs, distances)
-	for _, b := range nodesResp.Enrs {
+
+	return nodes, nil
+}
+
+func (p *PortalProtocol) filterNodes(target *enode.Node, enrs [][]byte, distances []uint) []*enode.Node {
+	var (
+		nodes    []*enode.Node
+		seen     = make(map[enode.ID]struct{})
+		err      error
+		verified = 0
+		n        *enode.Node
+	)
+
+	for _, b := range enrs {
 		record := &enr.Record{}
 		err = rlp.DecodeBytes(b, record)
 		if err != nil {
-			p.log.Debug("Invalid record in nodes response", "id", target.ID(), "err", err)
+			p.log.Error("Invalid record in nodes response", "id", target.ID(), "err", err)
 			continue
 		}
 		n, err = p.verifyResponseNode(target, record, distances, seen)
 		if err != nil {
-			p.log.Debug("Invalid record in nodes response", "id", target.ID(), "err", err)
+			p.log.Error("Invalid record in nodes response", "id", target.ID(), "err", err)
 			continue
 		}
 		verified++
 		nodes = append(nodes, n)
 	}
 
-	p.log.Trace("Received nodes response", "id", target.ID(), "total", nodesResp.Total, "verified", verified, "nodes", nodes)
+	p.log.Trace("Received nodes response", "id", target.ID(), "total", len(enrs), "verified", verified, "nodes", nodes)
-	return nodes, nil
+	return nodes
 }
 
 func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (uint64, error) {
@@ -309,16 +455,16 @@ func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (uint64, e
 }
 
 func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte {
-	if node := p.DiscV5.getNode(id); node != nil {
+	if n := p.DiscV5.getNode(id); n != nil {
-		p.table.addSeenNode(wrapNode(node))
+		p.table.addSeenNode(wrapNode(n))
 	}
 	p.utpPackets <- &utp.UdpMessage{Buf: msg, Addr: addr}
 	return []byte("")
 }
 
 func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte {
-	if node := p.DiscV5.getNode(id); node != nil {
+	if n := p.DiscV5.getNode(id); n != nil {
-		p.table.addSeenNode(wrapNode(node))
+		p.table.addSeenNode(wrapNode(n))
 	}
 
 	msgCode := msg[0]
@@ -459,16 +605,16 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
 
 	contentId := p.storage.ContentId(request.ContentKey)
 	if contentId == nil {
-		return nil, fmt.Errorf("content not found")
+		return nil, ContentNotFound
 	}
 
 	var content []byte
 	content, err = p.storage.Get(request.ContentKey, contentId)
-	if err != nil {
+	if err != nil && !errors.Is(err, ContentNotFound) {
 		return nil, err
 	}
 
-	if content == nil {
+	if errors.Is(err, ContentNotFound) {
 		closestNodes := p.findNodesCloseToContent(contentId)
 		for i, n := range closestNodes {
 			if n.ID() == id {
@@ -500,9 +646,21 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
 
 		return talkRespBytes, nil
 	} else if len(content) <= maxPayloadSize {
-		contentMsgBytes := make([]byte, 0, len(content)+1)
+		rawContentMsg := &portalwire.Content{
+			Content: content,
+		}
+
+		p.log.Trace("Sending raw content response", "protocol", p.protocolId, "source", addr, "content", rawContentMsg)
+
+		var rawContentMsgBytes []byte
+		rawContentMsgBytes, err = rawContentMsg.MarshalSSZ()
+		if err != nil {
+			return nil, err
+		}
+
+		contentMsgBytes := make([]byte, 0, len(rawContentMsgBytes)+1)
 		contentMsgBytes = append(contentMsgBytes, portalwire.ContentRawSelector)
-		contentMsgBytes = append(contentMsgBytes, content...)
+		contentMsgBytes = append(contentMsgBytes, rawContentMsgBytes...)
 
 		talkRespBytes := make([]byte, 0, len(contentMsgBytes)+1)
 		talkRespBytes = append(talkRespBytes, portalwire.CONTENT)
@@ -515,7 +673,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
 		connIdSend := connId.SendId()
 
 		go func() {
-			ctx, cancel := context.WithTimeout(context.Background(), defaultUTPAcceptTimeout)
+			ctx, cancel := context.WithTimeout(context.Background(), defaultUTPConnectTimeout)
 			var conn *utp.Conn
 			conn, err = p.utp.AcceptUTPContext(ctx, connIdSend)
 			if err != nil {
@@ -604,44 +762,6 @@ func (p *PortalProtocol) verifyResponseNode(sender *enode.Node, r *enr.Record, d
 	return n, nil
 }
 
-func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) {
-	enrSeq := p.DiscV5.LocalNode().Seq()
-	radiusBytes, err := p.nodeRadius.MarshalSSZ()
-	if err != nil {
-		return 0, err
-	}
-	customPayload := &portalwire.PingPongCustomData{
-		Radius: radiusBytes,
-	}
-
-	customPayloadBytes, err := customPayload.MarshalSSZ()
-	if err != nil {
-		return 0, err
-	}
-
-	pingRequest := &portalwire.Ping{
-		EnrSeq:        enrSeq,
-		CustomPayload: customPayloadBytes,
-	}
-
-	p.log.Trace("Sending ping request", "protocol", p.protocolId, "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest)
-	pingRequestBytes, err := pingRequest.MarshalSSZ()
-	if err != nil {
-		return 0, err
-	}
-
-	talkRequestBytes := make([]byte, 0, len(pingRequestBytes)+1)
-	talkRequestBytes = append(talkRequestBytes, portalwire.PING)
-	talkRequestBytes = append(talkRequestBytes, pingRequestBytes...)
-
-	talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes)
-
-	if err != nil {
-		p.replaceNode(node)
-	}
-	return p.processPong(node, talkResp)
-}
-
 func (p *PortalProtocol) replaceNode(node *enode.Node) {
 	p.table.mutex.Lock()
 	defer p.table.mutex.Unlock()

p2p/discover/portal_protocol_test.go

@@ -1,10 +1,12 @@
 package discover
 
 import (
+	"crypto/rand"
 	"fmt"
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/internal/testlog"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover/portalwire"
@@ -13,6 +15,26 @@ import (
 	"golang.org/x/exp/slices"
 )
 
+type MockStorage struct {
+	db map[string][]byte
+}
+
+func (m *MockStorage) ContentId(contentKey []byte) []byte {
+	return crypto.Keccak256(contentKey)
+}
+
+func (m *MockStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) {
+	if content, ok := m.db[string(contentId)]; ok {
+		return content, nil
+	}
+	return nil, ContentNotFound
+}
+
+func (m *MockStorage) Put(contentKey []byte, content []byte) error {
+	m.db[string(m.ContentId(contentKey))] = content
+	return nil
+}
+
 func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol, error) {
 	conf := DefaultPortalProtocolConfig()
 	if addr != "" {
@@ -21,7 +43,7 @@ func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol
 	if bootNodes != nil {
 		conf.BootstrapNodes = bootNodes
 	}
-	portalProtocol, err := NewPortalProtocol(conf, portalwire.HistoryNetwork, newkey())
+	portalProtocol, err := NewPortalProtocol(conf, portalwire.HistoryNetwork, newkey(), &MockStorage{db: make(map[string][]byte)})
 	if err != nil {
 		return nil, err
 	}
@@ -76,4 +98,32 @@ func TestPortalWireProtocol(t *testing.T) {
 	slices.ContainsFunc(node3.table.Nodes(), func(n *enode.Node) bool {
 		return n.ID() == node2.localNode.Node().ID()
 	})
+
+	err = node1.storage.Put([]byte("test_key"), []byte("test_value"))
+	assert.NoError(t, err)
+
+	flag, content, err := node2.findContent(node1.localNode.Node(), []byte("test_key"))
+	assert.NoError(t, err)
+	assert.Equal(t, portalwire.ContentRawSelector, flag)
+	assert.Equal(t, []byte("test_value"), content)
+
+	flag, content, err = node2.findContent(node3.localNode.Node(), []byte("test_key"))
+	assert.NoError(t, err)
+	assert.Equal(t, portalwire.ContentEnrsSelector, flag)
+	assert.Equal(t, 1, len(content.([]*enode.Node)))
+	assert.Equal(t, node1.localNode.Node().ID(), content.([]*enode.Node)[0].ID())
+
+	// create a byte slice of length 1199 and fill it with random data
+	// this will be used as a test content
+	largeTestContent := make([]byte, 1199)
+	_, err = rand.Read(largeTestContent)
+	assert.NoError(t, err)
+
+	err = node1.storage.Put([]byte("large_test_key"), largeTestContent)
+	assert.NoError(t, err)
+
+	//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)
 }

p2p/discover/portal_storage.go

@@ -1,5 +1,9 @@
 package discover
 
+import "fmt"
+
+var ContentNotFound = fmt.Errorf("content not found")
+
 type Storage interface {
 	ContentId(contentKey []byte) []byte