forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-01 09:33:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat:beacon network test

Browse source 
Signed-off-by: Chen Kai <281165273grape@gmail.com>
This commit is contained in:
Chen Kai 2024-04-13 21:21:43 +08:00
parent 5cc962afc7
commit eb3206b1e1
9 changed files with 468 additions and 249 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
p2p/discover/api.go
View file

@ -384,14 +384,14 @@ func (p *PortalProtocolAPI) FindContent(enr string, contentKey string) (interfac
Content: hexutil.Encode(findContent.([]byte)),
UtpTransfer: false,
}
p.portalProtocol.log.Trace("FindContent", "contentInfo", contentInfo)
p.portalProtocol.Log.Trace("FindContent", "contentInfo", contentInfo)
return contentInfo, nil
case portalwire.ContentConnIdSelector:
contentInfo := &ContentInfo{
Content: hexutil.Encode(findContent.([]byte)),
UtpTransfer: true,
}
p.portalProtocol.log.Trace("FindContent", "contentInfo", contentInfo)
p.portalProtocol.Log.Trace("FindContent", "contentInfo", contentInfo)
return contentInfo, nil
default:
enrs := make([]string, 0)
@ -399,7 +399,7 @@ func (p *PortalProtocolAPI) FindContent(enr string, contentKey string) (interfac
enrs = append(enrs, r.String())
}
p.portalProtocol.log.Trace("FindContent", "enrs", enrs)
p.portalProtocol.Log.Trace("FindContent", "enrs", enrs)
return &Enrs{
Enrs: enrs,
}, nil
@ -515,7 +515,7 @@ func (p *PortalProtocolAPI) Gossip(contentKeyHex, contentHex string) (int, error
return 0, err
}
id := p.portalProtocol.Self().ID()
return p.portalProtocol.NeighborhoodGossip(&id, [][]byte{contentKey}, [][]byte{content})
return p.portalProtocol.Gossip(&id, [][]byte{contentKey}, [][]byte{content})
}
func (p *PortalProtocolAPI) TraceRecursiveFindContent(contentKeyHex string) (*TraceContentResult, error) {

260
p2p/discover/portal_protocol.go
View file

@ -165,7 +165,7 @@ type PortalProtocol struct {
packetRouter *utp.PacketRouter
ListenAddr string
localNode *enode.LocalNode
log log.Logger
Log log.Logger
PrivateKey *ecdsa.PrivateKey
NetRestrict *netutil.Netlist
BootstrapNodes []*enode.Node
@ -193,7 +193,7 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateK
protocol := &PortalProtocol{
protocolId: protocolId,
ListenAddr: config.ListenAddr,
log: log.New("protocol", protocolId),
Log: log.New("protocol", protocolId),
PrivateKey: privateKey,
NetRestrict: config.NetRestrict,
BootstrapNodes: config.BootstrapNodes,
@ -244,7 +244,7 @@ func (p *PortalProtocol) Stop() {
p.DiscV5.Close()
err := p.utp.Close()
if err != nil {
p.log.Error("failed to close utp listener", "err", err)
p.Log.Error("failed to close utp listener", "err", err)
}
}
func (p *PortalProtocol) RoutingTableInfo() [][]string {
@ -258,14 +258,14 @@ func (p *PortalProtocol) RoutingTableInfo() [][]string {
}
nodes = append(nodes, bucketNodes)
}
p.log.Trace("routingTableInfo resp:", "nodes", nodes)
p.Log.Trace("routingTableInfo resp:", "nodes", nodes)
return nodes
}
func (p *PortalProtocol) setupUDPListening() error {
laddr := p.conn.LocalAddr().(*net.UDPAddr)
p.localNode.SetFallbackUDP(laddr.Port)
p.log.Debug("UDP listener up", "addr", laddr)
p.Log.Debug("UDP listener up", "addr", laddr)
// TODO: NAT
//if !laddr.IP.IsLoopback() && !laddr.IP.IsPrivate() {
// srv.portMappingRegister <- &portMapping{
@ -298,14 +298,14 @@ func (p *PortalProtocol) setupUDPListening() error {
}
}
p.log.Trace("send to target data", "ip", target.IP().String(), "port", target.UDP(), "bufLength", len(buf))
p.Log.Trace("send to target data", "ip", target.IP().String(), "port", target.UDP(), "bufLength", len(buf))
_, err := p.DiscV5.TalkRequest(target, string(portalwire.UTPNetwork), buf)
return len(buf), err
})
ctx := context.Background()
var logger *zap.Logger
if p.log.Enabled(ctx, log.LevelDebug) || p.log.Enabled(ctx, log.LevelTrace) {
if p.Log.Enabled(ctx, log.LevelDebug) || p.Log.Enabled(ctx, log.LevelTrace) {
logger, err = zap.NewDevelopmentConfig().Build()
} else {
logger, err = zap.NewProductionConfig().Build()
@ -336,7 +336,7 @@ func (p *PortalProtocol) setupDiscV5AndTable() error {
PrivateKey: p.PrivateKey,
NetRestrict: p.NetRestrict,
Bootnodes: p.BootstrapNodes,
Log: p.log,
Log: p.Log,
}
p.table, err = newMeteredTable(p, p.localNode.Database(), cfg)
@ -376,7 +376,7 @@ func (p *PortalProtocol) pingInner(node *enode.Node) (*portalwire.Pong, error) {
CustomPayload: customPayloadBytes,
}
p.log.Trace("Sending ping request", "protocol", p.protocolId, "ip", p.Self().IP().String(), "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest)
p.Log.Trace("Sending ping request", "protocol", p.protocolId, "ip", p.Self().IP().String(), "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest)
pingRequestBytes, err := pingRequest.MarshalSSZ()
if err != nil {
return nil, err
@ -389,12 +389,12 @@ func (p *PortalProtocol) pingInner(node *enode.Node) (*portalwire.Pong, error) {
talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes)
if err != nil {
p.log.Error("ping error:", err)
p.Log.Error("ping error:", err)
p.replaceNode(node)
return nil, err
}
p.log.Trace("Received ping response", "source", p.Self().ID(), "target", node.ID(), "res", talkResp)
p.Log.Trace("Received ping response", "source", p.Self().ID(), "target", node.ID(), "res", talkResp)
return p.processPong(node, talkResp)
}
@ -409,10 +409,10 @@ func (p *PortalProtocol) findNodes(node *enode.Node, distances []uint) ([]*enode
Distances: distancesBytes,
}
p.log.Trace("Sending find nodes request", "id", node.ID(), "findNodes", findNodes)
p.Log.Trace("Sending find nodes request", "id", node.ID(), "findNodes", findNodes)
findNodesBytes, err := findNodes.MarshalSSZ()
if err != nil {
p.log.Error("failed to marshal find nodes request", "err", err)
p.Log.Error("failed to marshal find nodes request", "err", err)
return nil, err
}
@ -422,7 +422,7 @@ func (p *PortalProtocol) findNodes(node *enode.Node, distances []uint) ([]*enode
talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes)
if err != nil {
p.log.Error("failed to send find nodes request", "err", err)
p.Log.Error("failed to send find nodes request", "err", err)
return nil, err
}
@ -434,10 +434,10 @@ func (p *PortalProtocol) findContent(node *enode.Node, contentKey []byte) (byte,
ContentKey: contentKey,
}
p.log.Trace("Sending find content request", "id", node.ID(), "findContent", findContent)
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)
p.Log.Error("failed to marshal find content request", "err", err)
return 0xff, nil, err
}
@ -447,7 +447,7 @@ func (p *PortalProtocol) findContent(node *enode.Node, contentKey []byte) (byte,
talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes)
if err != nil {
p.log.Error("failed to send find content request", "err", err)
p.Log.Error("failed to send find content request", "err", err)
return 0xff, nil, err
}
@ -461,10 +461,10 @@ func (p *PortalProtocol) offer(node *enode.Node, offerRequest *OfferRequest) ([]
ContentKeys: contentKeys,
}
p.log.Trace("Sending offer request", "offer", offer)
p.Log.Trace("Sending offer request", "offer", offer)
offerBytes, err := offer.MarshalSSZ()
if err != nil {
p.log.Error("failed to marshal offer request", "err", err)
p.Log.Error("failed to marshal offer request", "err", err)
return nil, err
}
@ -474,7 +474,7 @@ func (p *PortalProtocol) offer(node *enode.Node, offerRequest *OfferRequest) ([]
talkResp, err := p.DiscV5.TalkRequest(node, p.protocolId, talkRequestBytes)
if err != nil {
p.log.Error("failed to send offer request", "err", err)
p.Log.Error("failed to send offer request", "err", err)
return nil, err
}
@ -493,7 +493,7 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
return nil, err
}
p.log.Trace("Received accept response", "id", target.ID(), "accept", accept)
p.Log.Trace("Received accept response", "id", target.ID(), "accept", accept)
p.setJustSeen(target)
var contentKeyLen int
@ -534,7 +534,7 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
if contentId != nil {
content, err = p.storage.Get(contentKey, contentId)
if err != nil {
p.log.Error("failed to get content from storage", "err", err)
p.Log.Error("failed to get content from storage", "err", err)
contents = append(contents, []byte{})
} else {
contents = append(contents, content)
@ -548,7 +548,7 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
var contentsPayload []byte
contentsPayload, err = encodeContents(contents)
if err != nil {
p.log.Error("failed to encode contents", "err", err)
p.Log.Error("failed to encode contents", "err", err)
return
}
@ -559,17 +559,17 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
if err != nil {
conncancel()
p.log.Error("failed to dial utp connection", "err", err)
p.Log.Error("failed to dial utp connection", "err", err)
return
}
conncancel()
err = conn.SetWriteDeadline(time.Now().Add(defaultUTPWriteTimeout))
if err != nil {
p.log.Error("failed to set write deadline", "err", err)
p.Log.Error("failed to set write deadline", "err", err)
err = conn.Close()
if err != nil {
p.log.Error("failed to close utp connection", "err", err)
p.Log.Error("failed to close utp connection", "err", err)
return
}
@ -579,18 +579,18 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
var written int
written, err = conn.Write(contentsPayload)
if err != nil {
p.log.Error("failed to write to utp connection", "err", err)
p.Log.Error("failed to write to utp connection", "err", err)
err = conn.Close()
if err != nil {
p.log.Error("failed to close utp connection", "err", err)
p.Log.Error("failed to close utp connection", "err", err)
return
}
return
}
p.log.Trace("Sent content response", "id", target.ID(), "contents", contents, "size", written)
p.Log.Trace("Sent content response", "id", target.ID(), "contents", contents, "size", written)
err = conn.Close()
if err != nil {
p.log.Error("failed to close utp connection", "err", err)
p.Log.Error("failed to close utp connection", "err", err)
return
}
return
@ -614,7 +614,7 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
return 0xff, nil, err
}
p.log.Trace("Received content response", "id", target.ID(), "content", content)
p.Log.Trace("Received content response", "id", target.ID(), "content", content)
p.setJustSeen(target)
return resp[1], content.Content, nil
case portalwire.ContentConnIdSelector:
@ -624,7 +624,7 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
return 0xff, nil, err
}
p.log.Trace("Received returned content response", "id", target.ID(), "connIdMsg", connIdMsg)
p.Log.Trace("Received returned content response", "id", target.ID(), "connIdMsg", connIdMsg)
p.setJustSeen(target)
connctx, conncancel := context.WithTimeout(p.closeCtx, defaultUTPConnectTimeout)
laddr := p.utp.Addr().(*utp.Addr)
@ -644,10 +644,10 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
// Read ALL the data from the connection until EOF and return it
data, err := io.ReadAll(conn)
if err != nil {
p.log.Error("failed to read from utp connection", "err", err)
p.Log.Error("failed to read from utp connection", "err", err)
return 0xff, nil, err
}
p.log.Trace("Received content response", "id", target.ID(), "size", len(data), "data", data)
p.Log.Trace("Received content response", "id", target.ID(), "size", len(data), "data", data)
return resp[1], data, nil
case portalwire.ContentEnrsSelector:
enrs := &portalwire.Enrs{}
@ -657,7 +657,7 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
return 0xff, nil, err
}
p.log.Trace("Received content response", "id", target.ID(), "enrs", enrs)
p.Log.Trace("Received content response", "id", target.ID(), "enrs", enrs)
p.setJustSeen(target)
nodes := p.filterNodes(target, enrs.Enrs, nil)
return resp[1], nodes, nil
@ -702,19 +702,19 @@ func (p *PortalProtocol) filterNodes(target *enode.Node, enrs [][]byte, distance
record := &enr.Record{}
err = rlp.DecodeBytes(b, record)
if err != nil {
p.log.Error("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.Error("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", len(enrs), "verified", verified, "nodes", nodes)
p.Log.Trace("Received nodes response", "id", target.ID(), "total", len(enrs), "verified", verified, "nodes", nodes)
return nodes
}
@ -729,7 +729,7 @@ func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (*portalwi
return nil, err
}
p.log.Trace("Received pong response", "id", target.ID(), "pong", pong)
p.Log.Trace("Received pong response", "id", target.ID(), "pong", pong)
customPayload := &portalwire.PingPongCustomData{}
err = customPayload.UnmarshalSSZ(pong.CustomPayload)
@ -738,7 +738,7 @@ func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (*portalwi
return nil, err
}
p.log.Trace("Received pong response", "id", target.ID(), "pong", pong, "customPayload", customPayload)
p.Log.Trace("Received pong response", "id", target.ID(), "pong", pong, "customPayload", customPayload)
p.setJustSeen(target)
p.radiusCache.Set([]byte(target.ID().String()), customPayload.Radius)
@ -749,13 +749,13 @@ func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, ms
if n := p.DiscV5.getNode(id); n != nil {
p.table.addSeenNode(wrapNode(n))
}
p.log.Trace("receive utp data", "addr", addr, "msg-length", len(msg))
p.Log.Trace("receive utp data", "addr", addr, "msg-length", len(msg))
p.packetRouter.ReceiveMessage(msg, addr)
return []byte("")
}
func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte {
p.log.Trace("handleTalkRequest", "id", id, "addr", addr)
p.Log.Trace("handleTalkRequest", "id", id, "addr", addr)
if n := p.DiscV5.getNode(id); n != nil {
p.table.addSeenNode(wrapNode(n))
}
@ -767,14 +767,14 @@ func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg [
pingRequest := &portalwire.Ping{}
err := pingRequest.UnmarshalSSZ(msg[1:])
if err != nil {
p.log.Error("failed to unmarshal ping request", "err", err)
p.Log.Error("failed to unmarshal ping request", "err", err)
return nil
}
p.log.Trace("received ping request", "protocol", p.protocolId, "source", id, "pingRequest", pingRequest)
p.Log.Trace("received ping request", "protocol", p.protocolId, "source", id, "pingRequest", pingRequest)
resp, err := p.handlePing(id, pingRequest)
if err != nil {
p.log.Error("failed to handle ping request", "err", err)
p.Log.Error("failed to handle ping request", "err", err)
return nil
}
@ -783,14 +783,14 @@ func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg [
findNodesRequest := &portalwire.FindNodes{}
err := findNodesRequest.UnmarshalSSZ(msg[1:])
if err != nil {
p.log.Error("failed to unmarshal find nodes request", "err", err)
p.Log.Error("failed to unmarshal find nodes request", "err", err)
return nil
}
p.log.Trace("received find nodes request", "protocol", p.protocolId, "source", id, "findNodesRequest", findNodesRequest)
p.Log.Trace("received find nodes request", "protocol", p.protocolId, "source", id, "findNodesRequest", findNodesRequest)
resp, err := p.handleFindNodes(addr, findNodesRequest)
if err != nil {
p.log.Error("failed to handle find nodes request", "err", err)
p.Log.Error("failed to handle find nodes request", "err", err)
return nil
}
@ -799,14 +799,14 @@ func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg [
findContentRequest := &portalwire.FindContent{}
err := findContentRequest.UnmarshalSSZ(msg[1:])
if err != nil {
p.log.Error("failed to unmarshal find content request", "err", err)
p.Log.Error("failed to unmarshal find content request", "err", err)
return nil
}
p.log.Trace("received find content request", "protocol", p.protocolId, "source", id, "findContentRequest", findContentRequest)
p.Log.Trace("received find content request", "protocol", p.protocolId, "source", id, "findContentRequest", findContentRequest)
resp, err := p.handleFindContent(id, addr, findContentRequest)
if err != nil {
p.log.Error("failed to handle find content request", "err", err)
p.Log.Error("failed to handle find content request", "err", err)
return nil
}
@ -815,14 +815,14 @@ func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg [
offerRequest := &portalwire.Offer{}
err := offerRequest.UnmarshalSSZ(msg[1:])
if err != nil {
p.log.Error("failed to unmarshal offer request", "err", err)
p.Log.Error("failed to unmarshal offer request", "err", err)
return nil
}
p.log.Trace("received offer request", "protocol", p.protocolId, "source", id, "offerRequest", offerRequest)
p.Log.Trace("received offer request", "protocol", p.protocolId, "source", id, "offerRequest", offerRequest)
resp, err := p.handleOffer(id, addr, offerRequest)
if err != nil {
p.log.Error("failed to handle offer request", "err", err)
p.Log.Error("failed to handle offer request", "err", err)
return nil
}
@ -860,7 +860,7 @@ func (p *PortalProtocol) handlePing(id enode.ID, ping *portalwire.Ping) ([]byte,
CustomPayload: pongCustomPayloadBytes,
}
p.log.Trace("Sending pong response", "protocol", p.protocolId, "source", id, "pong", pong)
p.Log.Trace("Sending pong response", "protocol", p.protocolId, "source", id, "pong", pong)
pongBytes, err := pong.MarshalSSZ()
if err != nil {
@ -893,7 +893,7 @@ func (p *PortalProtocol) handleFindNodes(fromAddr *net.UDPAddr, request *portalw
Enrs: enrs,
}
p.log.Trace("Sending nodes response", "protocol", p.protocolId, "source", fromAddr, "nodes", nodesMsg)
p.Log.Trace("Sending nodes response", "protocol", p.protocolId, "source", fromAddr, "nodes", nodesMsg)
nodesMsgBytes, err := nodesMsg.MarshalSSZ()
if err != nil {
return nil, err
@ -942,7 +942,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
Enrs: enrs,
}
p.log.Trace("Sending enrs content response", "protocol", p.protocolId, "source", addr, "enrs", enrsMsg)
p.Log.Trace("Sending enrs content response", "protocol", p.protocolId, "source", addr, "enrs", enrsMsg)
var enrsMsgBytes []byte
enrsMsgBytes, err = enrsMsg.MarshalSSZ()
if err != nil {
@ -963,7 +963,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
Content: content,
}
p.log.Trace("Sending raw content response", "protocol", p.protocolId, "source", addr, "content", rawContentMsg)
p.Log.Trace("Sending raw content response", "protocol", p.protocolId, "source", addr, "content", rawContentMsg)
var rawContentMsgBytes []byte
rawContentMsgBytes, err = rawContentMsg.MarshalSSZ()
@ -995,7 +995,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
var conn *utp.Conn
conn, err = p.utp.AcceptUTPContext(ctx, connIdSend)
if err != nil {
p.log.Error("failed to accept utp connection", "connId", connIdSend, "err", err)
p.Log.Error("failed to accept utp connection", "connId", connIdSend, "err", err)
cancel()
return
}
@ -1003,10 +1003,10 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
err = conn.SetWriteDeadline(time.Now().Add(defaultUTPWriteTimeout))
if err != nil {
p.log.Error("failed to set write deadline", "err", err)
p.Log.Error("failed to set write deadline", "err", err)
err = conn.Close()
if err != nil {
p.log.Error("failed to close utp connection", "err", err)
p.Log.Error("failed to close utp connection", "err", err)
return
}
return
@ -1015,10 +1015,10 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
var n int
n, err = conn.Write(content)
if err != nil {
p.log.Error("failed to write content to utp connection", "err", err)
p.Log.Error("failed to write content to utp connection", "err", err)
err = conn.Close()
if err != nil {
p.log.Error("failed to close utp connection", "err", err)
p.Log.Error("failed to close utp connection", "err", err)
return
}
return
@ -1026,11 +1026,11 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
err = conn.Close()
if err != nil {
p.log.Error("failed to close utp connection", "err", err)
p.Log.Error("failed to close utp connection", "err", err)
return
}
p.log.Trace("wrote content size to utp connection", "n", n)
p.Log.Trace("wrote content size to utp connection", "n", n)
return
}
}
@ -1042,7 +1042,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
Id: idBuffer,
}
p.log.Trace("Sending connection id content response", "protocol", p.protocolId, "source", addr, "connId", connIdMsg)
p.Log.Trace("Sending connection id content response", "protocol", p.protocolId, "source", addr, "connId", connIdMsg)
var connIdMsgBytes []byte
connIdMsgBytes, err = connIdMsg.MarshalSSZ()
if err != nil {
@ -1070,7 +1070,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
ContentKeys: contentKeyBitlist,
}
p.log.Trace("Sending accept response", "protocol", p.protocolId, "source", addr, "accept", acceptMsg)
p.Log.Trace("Sending accept response", "protocol", p.protocolId, "source", addr, "accept", acceptMsg)
var acceptMsgBytes []byte
acceptMsgBytes, err = acceptMsg.MarshalSSZ()
if err != nil {
@ -1115,7 +1115,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
var conn *utp.Conn
conn, err = p.utp.AcceptUTPContext(ctx, connIdSend)
if err != nil {
p.log.Error("failed to accept utp connection", "connId", connIdSend, "err", err)
p.Log.Error("failed to accept utp connection", "connId", connIdSend, "err", err)
cancel()
return
}
@ -1123,21 +1123,21 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
err = conn.SetReadDeadline(time.Now().Add(defaultUTPReadTimeout))
if err != nil {
p.log.Error("failed to set read deadline", "err", err)
p.Log.Error("failed to set read deadline", "err", err)
return
}
// Read ALL the data from the connection until EOF and return it
var data []byte
data, err = io.ReadAll(conn)
if err != nil {
p.log.Error("failed to read from utp connection", "err", err)
p.Log.Error("failed to read from utp connection", "err", err)
return
}
p.log.Trace("Received offer content response", "id", id, "size", len(data), "data", data)
p.Log.Trace("Received offer content response", "id", id, "size", len(data), "data", data)
err = p.handleOfferedContents(id, contentKeys, data)
if err != nil {
p.log.Error("failed to handle offered Contents", "err", err)
p.Log.Error("failed to handle offered Contents", "err", err)
return
}
@ -1156,7 +1156,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
ContentKeys: []byte(contentKeyBitlist),
}
p.log.Trace("Sending accept response", "protocol", p.protocolId, "source", addr, "accept", acceptMsg)
p.Log.Trace("Sending accept response", "protocol", p.protocolId, "source", addr, "accept", acceptMsg)
var acceptMsgBytes []byte
acceptMsgBytes, err = acceptMsg.MarshalSSZ()
if err != nil {
@ -1295,10 +1295,10 @@ func (p *PortalProtocol) offerWorker() {
case <-p.closeCtx.Done():
return
case offerRequestWithNode := <-p.offerQueue:
p.log.Trace("offerWorker", "offerRequestWithNode", offerRequestWithNode)
p.Log.Trace("offerWorker", "offerRequestWithNode", offerRequestWithNode)
_, err := p.offer(offerRequestWithNode.Node, offerRequestWithNode.Request)
if err != nil {
p.log.Error("failed to offer", "err", err)
p.Log.Error("failed to offer", "err", err)
}
}
}
@ -1310,7 +1310,7 @@ func (p *PortalProtocol) truncateNodes(nodes []*enode.Node, maxSize int, enrOver
for _, n := range nodes {
enrBytes, err := rlp.EncodeToBytes(n.Record())
if err != nil {
p.log.Error("failed to encode n", "err", err)
p.Log.Error("failed to encode n", "err", err)
continue
}
@ -1619,21 +1619,21 @@ func (p *PortalProtocol) InRange(contentId []byte) bool {
func (p *PortalProtocol) Get(contentKey []byte, contentId []byte) ([]byte, error) {
content, err := p.storage.Get(contentKey, contentId)
p.log.Trace("get local storage", "contentId", hexutil.Encode(contentId), "content", hexutil.Encode(content), "err", err)
p.Log.Trace("get local storage", "contentId", hexutil.Encode(contentId), "content", hexutil.Encode(content), "err", err)
return content, err
}
func (p *PortalProtocol) Put(contentKey []byte, contentId []byte, content []byte) error {
err := p.storage.Put(contentKey, contentId, content)
p.log.Trace("put local storage", "contentId", hexutil.Encode(contentId), "content", hexutil.Encode(content), "err", err)
p.Log.Trace("put local storage", "contentId", hexutil.Encode(contentId), "content", hexutil.Encode(content), "err", err)
return err
}
func (p *PortalProtocol) GetContent() <-chan *ContentElement {
func (p *PortalProtocol) GetContent() chan *ContentElement {
return p.contentQueue
}
func (p *PortalProtocol) NeighborhoodGossip(srcNodeId *enode.ID, contentKeys [][]byte, content [][]byte) (int, error) {
func (p *PortalProtocol) Gossip(srcNodeId *enode.ID, contentKeys [][]byte, content [][]byte) (int, error) {
if len(content) == 0 {
return 0, errors.New("empty content")
}
@ -1652,23 +1652,9 @@ func (p *PortalProtocol) NeighborhoodGossip(srcNodeId *enode.ID, contentKeys [][
return 0, ErrNilContentKey
}
// For selecting the closest nodes to whom to gossip the content a mixed
// approach is taken:
// 1. Select the closest neighbours in the routing table
// 2. Check if the radius is known for these these nodes and whether they are
// in range of the content to be offered.
// 3. If more than n (= 8) nodes are in range, offer these nodes the content
// (max nodes set at 8).
// 4. If less than n nodes are in range, do a node lookup, and offer the nodes
// returned from the lookup the content (max nodes set at 8)
//
// This should give a bigger rate of success and avoid the data being stopped
// in its propagation than when looking only for nodes in the own routing
// table, but at the same time avoid unnecessary node lookups.
// It might still cause issues in data getting propagated in a wider id range.
maxGossipNodes := 8
closestLocalNodes := p.findNodesCloseToContent(contentId, 16)
maxClosestNodes := 4
maxFartherNodes := 4
closestLocalNodes := p.findNodesCloseToContent(contentId, 32)
gossipNodes := make([]*enode.Node, 0)
for _, n := range closestLocalNodes {
@ -1689,74 +1675,22 @@ func (p *PortalProtocol) NeighborhoodGossip(srcNodeId *enode.ID, contentKeys [][
}
}
if len(gossipNodes) >= maxGossipNodes {
numberOfGossipedNodes := min(len(gossipNodes), maxGossipNodes)
for _, n := range gossipNodes[:numberOfGossipedNodes] {
transientOfferRequest := &TransientOfferRequest{
Contents: contentList,
}
if len(gossipNodes) == 0 {
return 0, nil
}
offerRequest := &OfferRequest{
Kind: TransientOfferRequestKind,
Request: transientOfferRequest,
}
offerRequestWithNode := &OfferRequestWithNode{
Node: n,
Request: offerRequest,
}
p.offerQueue <- offerRequestWithNode
}
return numberOfGossipedNodes, nil
var finalGossipNodes []*enode.Node
if len(gossipNodes) > maxClosestNodes {
fartherNodes := gossipNodes[maxClosestNodes:]
rand.Shuffle(len(fartherNodes), func(i, j int) {
fartherNodes[i], fartherNodes[j] = fartherNodes[j], fartherNodes[i]
})
finalGossipNodes = append(gossipNodes[:maxClosestNodes], fartherNodes[:min(maxFartherNodes, len(fartherNodes))]...)
} else {
closestNodes := p.Lookup(enode.ID(contentId))
numberOfGossipedNodes := min(len(closestNodes), maxGossipNodes)
// Note: opportunistically not checking if the radius of the node is known
// and thus if the node is in radius with the content. Reason is, these
// should really be the closest nodes in the DHT, and thus are most likely
// going to be in range of the requested content.
for _, n := range closestNodes[:numberOfGossipedNodes] {
transientOfferRequest := &TransientOfferRequest{
Contents: contentList,
}
offerRequest := &OfferRequest{
Kind: TransientOfferRequestKind,
Request: transientOfferRequest,
}
offerRequestWithNode := &OfferRequestWithNode{
Node: n,
Request: offerRequest,
}
p.offerQueue <- offerRequestWithNode
}
return numberOfGossipedNodes, nil
}
}
func (p *PortalProtocol) RandomGossip(srcNodeId *enode.ID, contentKeys [][]byte, content [][]byte) (int, error) {
if len(content) == 0 {
return 0, errors.New("empty content")
finalGossipNodes = gossipNodes
}
contentList := make([]*ContentEntry, 0, portalwire.ContentKeysLimit)
for i := 0; i < len(content); i++ {
contentEntry := &ContentEntry{
ContentKey: contentKeys[i],
Content: content[i],
}
contentList = append(contentList, contentEntry)
}
maxGossipNodes := 4
nodes := RandomNodes(p.table, maxGossipNodes, func(node *enode.Node) bool {
return srcNodeId == nil || node.ID() != *srcNodeId
})
for _, n := range nodes {
for _, n := range finalGossipNodes {
transientOfferRequest := &TransientOfferRequest{
Contents: contentList,
}
@ -1773,7 +1707,7 @@ func (p *PortalProtocol) RandomGossip(srcNodeId *enode.ID, contentKeys [][]byte,
p.offerQueue <- offerRequestWithNode
}
return len(nodes), nil
return len(finalGossipNodes), nil
}
func (p *PortalProtocol) Distance(a, b enode.ID) enode.ID {
@ -1781,7 +1715,7 @@ func (p *PortalProtocol) Distance(a, b enode.ID) enode.ID {
for i := range a {
res[i] = a[i] ^ b[i]
}
return enode.ID(res)
return res
}
func inRange(nodeId enode.ID, nodeRadius *uint256.Int, contentId []byte) bool {

65
p2p/discover/portal_protocol_test.go
View file

@ -11,6 +11,7 @@ import (
"testing"
"time"
"github.com/ethereum/go-ethereum/portalnetwork/storage"
"github.com/prysmaticlabs/go-bitfield"
"golang.org/x/exp/slices"
@ -23,22 +24,6 @@ import (
"github.com/stretchr/testify/assert"
)
type MockStorage struct {
db map[string][]byte
}
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, contentId []byte, content []byte) error {
m.db[string(contentId)] = content
return nil
}
func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol, error) {
conf := DefaultPortalProtocolConfig()
if addr != "" {
@ -101,7 +86,7 @@ func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol
}
contentQueue := make(chan *ContentElement, 50)
portalProtocol, err := NewPortalProtocol(conf, string(portalwire.HistoryNetwork), privKey, conn, localNode, discV5, &MockStorage{db: make(map[string][]byte)}, contentQueue)
portalProtocol, err := NewPortalProtocol(conf, string(portalwire.HistoryNetwork), privKey, conn, localNode, discV5, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue)
if err != nil {
return nil, err
}
@ -112,19 +97,19 @@ func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol
func TestPortalWireProtocolUdp(t *testing.T) {
node1, err := setupLocalPortalNode(":8777", nil)
assert.NoError(t, err)
node1.log = testlog.Logger(t, log.LvlTrace)
node1.Log = testlog.Logger(t, log.LvlTrace)
err = node1.Start()
assert.NoError(t, err)
node2, err := setupLocalPortalNode(":8778", []*enode.Node{node1.localNode.Node()})
assert.NoError(t, err)
node2.log = testlog.Logger(t, log.LvlTrace)
node2.Log = testlog.Logger(t, log.LvlTrace)
err = node2.Start()
assert.NoError(t, err)
node3, err := setupLocalPortalNode(":8779", []*enode.Node{node1.localNode.Node()})
assert.NoError(t, err)
node3.log = testlog.Logger(t, log.LvlTrace)
node3.Log = testlog.Logger(t, log.LvlTrace)
err = node3.Start()
assert.NoError(t, err)
time.Sleep(10 * time.Second)
@ -254,7 +239,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
func TestPortalWireProtocol(t *testing.T) {
node1, err := setupLocalPortalNode(":7777", nil)
assert.NoError(t, err)
node1.log = testlog.Logger(t, log.LevelDebug)
node1.Log = testlog.Logger(t, log.LevelDebug)
err = node1.Start()
assert.NoError(t, err)
fmt.Println(node1.localNode.Node().String())
@ -263,7 +248,7 @@ func TestPortalWireProtocol(t *testing.T) {
node2, err := setupLocalPortalNode(":7778", []*enode.Node{node1.localNode.Node()})
assert.NoError(t, err)
node2.log = testlog.Logger(t, log.LevelDebug)
node2.Log = testlog.Logger(t, log.LevelDebug)
err = node2.Start()
assert.NoError(t, err)
fmt.Println(node2.localNode.Node().String())
@ -272,7 +257,7 @@ func TestPortalWireProtocol(t *testing.T) {
node3, err := setupLocalPortalNode(":7779", []*enode.Node{node1.localNode.Node()})
assert.NoError(t, err)
node3.log = testlog.Logger(t, log.LevelDebug)
node3.Log = testlog.Logger(t, log.LevelDebug)
err = node3.Start()
assert.NoError(t, err)
fmt.Println(node3.localNode.Node().String())
@ -365,7 +350,7 @@ func TestPortalWireProtocol(t *testing.T) {
testGossipContentKeys := [][]byte{[]byte("test_gossip_content_keys"), []byte("test_gossip_content_keys2")}
testGossipContent := [][]byte{[]byte("test_gossip_content"), []byte("test_gossip_content2")}
id := node1.Self().ID()
gossip, err := node1.NeighborhoodGossip(&id, testGossipContentKeys, testGossipContent)
gossip, err := node1.Gossip(&id, testGossipContentKeys, testGossipContent)
assert.NoError(t, err)
assert.Equal(t, 2, gossip)
@ -383,26 +368,6 @@ func TestPortalWireProtocol(t *testing.T) {
assert.Equal(t, testGossipContentKeys[1], contentElement.ContentKeys[1])
assert.Equal(t, testGossipContent[1], contentElement.Contents[1])
testRandGossipContentKeys := [][]byte{[]byte("test_rand_gossip_content_keys"), []byte("test_rand_gossip_content_keys2")}
testRandGossipContent := [][]byte{[]byte("test_rand_gossip_content"), []byte("test_rand_gossip_content2")}
randGossip, err := node1.RandomGossip(&id, testRandGossipContentKeys, testRandGossipContent)
assert.NoError(t, err)
assert.Equal(t, 2, randGossip)
contentElement = <-node2.contentQueue
assert.Equal(t, node1.localNode.Node().ID(), contentElement.Node)
assert.Equal(t, testRandGossipContentKeys[0], contentElement.ContentKeys[0])
assert.Equal(t, testRandGossipContent[0], contentElement.Contents[0])
assert.Equal(t, testRandGossipContentKeys[1], contentElement.ContentKeys[1])
assert.Equal(t, testRandGossipContent[1], contentElement.Contents[1])
contentElement = <-node3.contentQueue
assert.Equal(t, node1.localNode.Node().ID(), contentElement.Node)
assert.Equal(t, testRandGossipContentKeys[0], contentElement.ContentKeys[0])
assert.Equal(t, testRandGossipContent[0], contentElement.Contents[0])
assert.Equal(t, testRandGossipContentKeys[1], contentElement.ContentKeys[1])
assert.Equal(t, testRandGossipContent[1], contentElement.Contents[1])
node1.Stop()
node2.Stop()
node3.Stop()
@ -428,21 +393,21 @@ func TestCancel(t *testing.T) {
func TestContentLookup(t *testing.T) {
node1, err := setupLocalPortalNode(":17777", nil)
assert.NoError(t, err)
node1.log = testlog.Logger(t, log.LvlTrace)
node1.Log = testlog.Logger(t, log.LvlTrace)
err = node1.Start()
assert.NoError(t, err)
fmt.Println(node1.localNode.Node().String())
node2, err := setupLocalPortalNode(":17778", []*enode.Node{node1.localNode.Node()})
assert.NoError(t, err)
node2.log = testlog.Logger(t, log.LvlTrace)
node2.Log = testlog.Logger(t, log.LvlTrace)
err = node2.Start()
assert.NoError(t, err)
fmt.Println(node2.localNode.Node().String())
node3, err := setupLocalPortalNode(":17779", []*enode.Node{node1.localNode.Node()})
assert.NoError(t, err)
node3.log = testlog.Logger(t, log.LvlTrace)
node3.Log = testlog.Logger(t, log.LvlTrace)
err = node3.Start()
assert.NoError(t, err)
fmt.Println(node3.localNode.Node().String())
@ -468,19 +433,19 @@ func TestContentLookup(t *testing.T) {
func TestTraceContentLookup(t *testing.T) {
node1, err := setupLocalPortalNode(":17787", nil)
assert.NoError(t, err)
node1.log = testlog.Logger(t, log.LvlTrace)
node1.Log = testlog.Logger(t, log.LvlTrace)
err = node1.Start()
assert.NoError(t, err)
node2, err := setupLocalPortalNode(":17788", []*enode.Node{node1.localNode.Node()})
assert.NoError(t, err)
node2.log = testlog.Logger(t, log.LvlTrace)
node2.Log = testlog.Logger(t, log.LvlTrace)
err = node2.Start()
assert.NoError(t, err)
node3, err := setupLocalPortalNode(":17789", []*enode.Node{node2.localNode.Node()})
assert.NoError(t, err)
node3.log = testlog.Logger(t, log.LvlTrace)
node3.Log = testlog.Logger(t, log.LvlTrace)
err = node3.Start()
assert.NoError(t, err)

118
portalnetwork/beacon/beacon_network.go
View file

@ -2,13 +2,18 @@ package beacon
import (
"bytes"
"context"
"errors"
"fmt"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/portalnetwork/storage"
ssz "github.com/ferranbt/fastssz"
"github.com/protolambda/zrnt/eth2/beacon/capella"
"github.com/protolambda/zrnt/eth2/beacon/common"
"github.com/protolambda/zrnt/eth2/configs"
"github.com/protolambda/ztyp/codec"
"github.com/protolambda/ztyp/tree"
)
@ -23,8 +28,38 @@ const (
)
type BeaconNetwork struct {
PortalProtocol *discover.PortalProtocol
Spec *common.Spec
portalProtocol *discover.PortalProtocol
spec *common.Spec
log log.Logger
closeCtx context.Context
closeFunc context.CancelFunc
}
func NewBeaconNetwork(portalProtocol *discover.PortalProtocol) *BeaconNetwork {
ctx, cancel := context.WithCancel(context.Background())
return &BeaconNetwork{
portalProtocol: portalProtocol,
spec: configs.Mainnet,
closeCtx: ctx,
closeFunc: cancel,
log: log.New("sub-protocol", "beacon"),
}
}
func (bn *BeaconNetwork) Start() error {
err := bn.portalProtocol.Start()
if err != nil {
return err
}
go bn.processContentLoop(bn.closeCtx)
bn.log.Debug("beacon network start successfully")
return nil
}
func (bn *BeaconNetwork) Stop() {
bn.closeFunc()
bn.portalProtocol.Stop()
}
func (bn *BeaconNetwork) GetUpdates(firstPeriod, count uint64) ([]*capella.LightClientUpdate, error) {
@ -39,7 +74,7 @@ func (bn *BeaconNetwork) GetUpdates(firstPeriod, count uint64) ([]*capella.Light
}
var lightClientUpdateRange LightClientUpdateRange = make([]ForkedLightClientUpdate, 0)
err = lightClientUpdateRange.Deserialize(bn.Spec, codec.NewDecodingReader(bytes.NewReader(lightClientUpdateRangeContent), uint64(len(lightClientUpdateRangeContent))))
err = lightClientUpdateRange.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(lightClientUpdateRangeContent), uint64(len(lightClientUpdateRangeContent))))
if err != nil {
return nil, err
}
@ -65,7 +100,7 @@ func (bn *BeaconNetwork) GetCheckpointData(checkpointHash tree.Root) (*capella.L
}
var forkedLightClientBootstrap ForkedLightClientBootstrap
err = forkedLightClientBootstrap.Deserialize(bn.Spec, codec.NewDecodingReader(bytes.NewReader(bootstrapValue), uint64(len(bootstrapValue))))
err = forkedLightClientBootstrap.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(bootstrapValue), uint64(len(bootstrapValue))))
if err != nil {
return nil, err
}
@ -88,7 +123,7 @@ func (bn *BeaconNetwork) GetFinalityUpdate(finalizedSlot uint64) (*capella.Light
}
var forkedLightClientFinalityUpdate ForkedLightClientFinalityUpdate
err = forkedLightClientFinalityUpdate.Deserialize(bn.Spec, codec.NewDecodingReader(bytes.NewReader(finalityUpdateValue), uint64(len(finalityUpdateValue))))
err = forkedLightClientFinalityUpdate.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(finalityUpdateValue), uint64(len(finalityUpdateValue))))
if err != nil {
return nil, err
}
@ -111,7 +146,7 @@ func (bn *BeaconNetwork) GetOptimisticUpdate(optimisticSlot uint64) (*capella.Li
}
var forkedLightClientOptimisticUpdate ForkedLightClientOptimisticUpdate
err = forkedLightClientOptimisticUpdate.Deserialize(bn.Spec, codec.NewDecodingReader(bytes.NewReader(optimisticUpdateValue), uint64(len(optimisticUpdateValue))))
err = forkedLightClientOptimisticUpdate.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(optimisticUpdateValue), uint64(len(optimisticUpdateValue))))
if err != nil {
return nil, err
}
@ -130,9 +165,9 @@ func (bn *BeaconNetwork) getContent(contentType storage.ContentType, beaconConte
}
contentKey := storage.NewContentKey(contentType, contentKeyBytes).Encode()
contentId := bn.PortalProtocol.ToContentId(contentKey)
contentId := bn.portalProtocol.ToContentId(contentKey)
res, err := bn.PortalProtocol.Get(contentKey, contentId)
res, err := bn.portalProtocol.Get(contentKey, contentId)
// other error
if err != nil && !errors.Is(err, storage.ErrContentNotFound) {
return nil, err
@ -142,10 +177,75 @@ func (bn *BeaconNetwork) getContent(contentType storage.ContentType, beaconConte
return res, nil
}
content, _, err := bn.PortalProtocol.ContentLookup(contentKey, contentId)
content, _, err := bn.portalProtocol.ContentLookup(contentKey, contentId)
if err != nil {
return nil, err
}
return content, nil
}
func (bn *BeaconNetwork) validateContent(contentKey []byte, content []byte) error {
switch storage.ContentType(contentKey[0]) {
case LightClientUpdate:
var lightClientUpdateRange LightClientUpdateRange = make([]ForkedLightClientUpdate, 0)
return lightClientUpdateRange.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(content), uint64(len(content))))
case LightClientBootstrap:
var forkedLightClientBootstrap ForkedLightClientBootstrap
return forkedLightClientBootstrap.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(content), uint64(len(content))))
// TODO: IF WE NEED LIGHT CLIENT VERIFY
case LightClientFinalityUpdate:
var forkedLightClientFinalityUpdate ForkedLightClientFinalityUpdate
return forkedLightClientFinalityUpdate.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(content), uint64(len(content))))
// TODO: IF WE NEED LIGHT CLIENT VERIFY
case LightClientOptimisticUpdate:
var forkedLightClientOptimisticUpdate ForkedLightClientOptimisticUpdate
return forkedLightClientOptimisticUpdate.Deserialize(bn.spec, codec.NewDecodingReader(bytes.NewReader(content), uint64(len(content))))
// TODO: VERIFY
case HistoricalSummaries:
var historicalSummaries HistoricalSummariesProof
return historicalSummaries.Deserialize(codec.NewDecodingReader(bytes.NewReader(content), uint64(len(content))))
default:
return fmt.Errorf("unknown content type %v", contentKey[0])
}
}
func (bn *BeaconNetwork) validateContents(contentKeys [][]byte, contents [][]byte) error {
for i, content := range contents {
contentKey := contentKeys[i]
err := bn.validateContent(contentKey, content)
if err != nil {
bn.log.Error("content validate failed", "contentKey", hexutil.Encode(contentKey), "content", hexutil.Encode(content), "err", err)
return fmt.Errorf("content validate failed with content key %x and content %x", contentKey, content)
}
contentId := bn.portalProtocol.ToContentId(contentKey)
err = bn.portalProtocol.Put(contentKey, contentId, content)
if err != nil {
bn.log.Error("put content failed", "contentKey", hexutil.Encode(contentKey), "content", hexutil.Encode(content), "err", err)
return err
}
}
return nil
}
func (bn *BeaconNetwork) processContentLoop(ctx context.Context) {
contentChan := bn.portalProtocol.GetContent()
for {
select {
case <-ctx.Done():
return
case contentElement := <-contentChan:
err := bn.validateContents(contentElement.ContentKeys, contentElement.Contents)
if err != nil {
bn.log.Error("validate content failed", "err", err)
continue
}
gossippedNum, err := bn.portalProtocol.Gossip(&contentElement.Node, contentElement.ContentKeys, contentElement.Contents)
bn.log.Trace("gossippedNum", "gossippedNum", gossippedNum)
if err != nil {
bn.log.Error("gossip failed", "err", err)
continue
}
}
}
}

220
portalnetwork/beacon/beacon_network_test.go Normal file
View file

@ -0,0 +1,220 @@
package beacon
import (
"encoding/json"
"net"
"os"
"testing"
"time"
"github.com/ethereum/go-ethereum/common/hexutil"
"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"
"github.com/ethereum/go-ethereum/p2p/discover/portalwire"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/portalnetwork/storage"
"github.com/stretchr/testify/assert"
)
func setupBeaconNetwork(addr string, bootNodes []*enode.Node) (*BeaconNetwork, error) {
conf := discover.DefaultPortalProtocolConfig()
if addr != "" {
conf.ListenAddr = addr
}
if bootNodes != nil {
conf.BootstrapNodes = bootNodes
}
addr1, err := net.ResolveUDPAddr("udp", conf.ListenAddr)
if err != nil {
return nil, err
}
conn, err := net.ListenUDP("udp", addr1)
if err != nil {
return nil, err
}
privKey, err := crypto.GenerateKey()
if err != nil {
panic("couldn't generate key: " + err.Error())
}
discCfg := discover.Config{
PrivateKey: privKey,
NetRestrict: conf.NetRestrict,
Bootnodes: conf.BootstrapNodes,
}
nodeDB, err := enode.OpenDB(conf.NodeDBPath)
if err != nil {
return nil, err
}
localNode := enode.NewLocalNode(nodeDB, privKey)
localNode.SetFallbackIP(net.IP{127, 0, 0, 1})
localNode.Set(discover.Tag)
var addrs []net.Addr
if conf.NodeIP != nil {
localNode.SetStaticIP(conf.NodeIP)
} else {
addrs, err = net.InterfaceAddrs()
if err != nil {
return nil, err
}
for _, address := range addrs {
// check ip addr is loopback addr
if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
if ipnet.IP.To4() != nil {
localNode.SetStaticIP(ipnet.IP)
break
}
}
}
}
discV5, err := discover.ListenV5(conn, localNode, discCfg)
if err != nil {
return nil, err
}
contentQueue := make(chan *discover.ContentElement, 50)
portalProtocol, err := discover.NewPortalProtocol(conf, string(portalwire.BeaconLightClientNetwork), privKey, conn, localNode, discV5, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue)
if err != nil {
return nil, err
}
return NewBeaconNetwork(portalProtocol), nil
}
func TestBeaconNetworkContent(t *testing.T) {
logger := testlog.Logger(t, log.LvlTrace)
node1, err := setupBeaconNetwork(":6666", nil)
assert.NoError(t, err)
node1.log = logger
node1.portalProtocol.Log = logger
err = node1.Start()
assert.NoError(t, err)
time.Sleep(10 * time.Second)
node2, err := setupBeaconNetwork(":6667", []*enode.Node{node1.portalProtocol.Self()})
assert.NoError(t, err)
node2.log = logger
node2.portalProtocol.Log = logger
err = node2.Start()
assert.NoError(t, err)
time.Sleep(10 * time.Second)
node3, err := setupBeaconNetwork(":6668", []*enode.Node{node1.portalProtocol.Self()})
assert.NoError(t, err)
node3.log = logger
node3.portalProtocol.Log = logger
err = node3.Start()
assert.NoError(t, err)
time.Sleep(10 * time.Second)
filePath := "testdata/light_client_updates_by_range.json"
jsonStr, _ := os.ReadFile(filePath)
var result map[string]interface{}
err = json.Unmarshal(jsonStr, &result)
assert.NoError(t, err)
id := node1.portalProtocol.Self().ID()
for _, v := range result {
kBytes, _ := hexutil.Decode(v.(map[string]interface{})["content_key"].(string))
vBytes, _ := hexutil.Decode(v.(map[string]interface{})["content_value"].(string))
contentKey := storage.NewContentKey(LightClientUpdate, kBytes).Encode()
_, err = node1.portalProtocol.Gossip(&id, [][]byte{contentKey}, [][]byte{vBytes})
assert.NoError(t, err)
time.Sleep(3 * time.Second)
contentId := node2.portalProtocol.ToContentId(contentKey)
get, err := node2.portalProtocol.Get(contentKey, contentId)
assert.NoError(t, err)
assert.Equal(t, vBytes, get)
}
filePath1 := "testdata/light_client_finality_update.json"
jsonStr1, _ := os.ReadFile(filePath1)
var result1 map[string]interface{}
err = json.Unmarshal(jsonStr1, &result1)
assert.NoError(t, err)
for _, v := range result1 {
kBytes1, _ := hexutil.Decode(v.(map[string]interface{})["content_key"].(string))
vBytes1, _ := hexutil.Decode(v.(map[string]interface{})["content_value"].(string))
contentKey1 := storage.NewContentKey(LightClientFinalityUpdate, kBytes1).Encode()
id = node1.portalProtocol.Self().ID()
_, err = node1.portalProtocol.Gossip(&id, [][]byte{contentKey1}, [][]byte{vBytes1})
assert.NoError(t, err)
time.Sleep(3 * time.Second)
contentId1 := node2.portalProtocol.ToContentId(contentKey1)
get1, err := node2.portalProtocol.Get(contentKey1, contentId1)
assert.NoError(t, err)
assert.Equal(t, vBytes1, get1)
}
filePath2 := "testdata/light_client_optimistic_update.json"
jsonStr2, _ := os.ReadFile(filePath2)
var result2 map[string]interface{}
err = json.Unmarshal(jsonStr2, &result2)
assert.NoError(t, err)
for _, v := range result2 {
kBytes2, _ := hexutil.Decode(v.(map[string]interface{})["content_key"].(string))
vBytes2, _ := hexutil.Decode(v.(map[string]interface{})["content_value"].(string))
contentKey2 := storage.NewContentKey(LightClientOptimisticUpdate, kBytes2).Encode()
id = node1.portalProtocol.Self().ID()
_, err = node1.portalProtocol.Gossip(&id, [][]byte{contentKey2}, [][]byte{vBytes2})
assert.NoError(t, err)
time.Sleep(3 * time.Second)
contentId2 := node2.portalProtocol.ToContentId(contentKey2)
get2, err := node2.portalProtocol.Get(contentKey2, contentId2)
assert.NoError(t, err)
assert.Equal(t, vBytes2, get2)
}
filePath3 := "testdata/light_client_bootstrap.json"
jsonStr3, _ := os.ReadFile(filePath3)
var result3 map[string]interface{}
err = json.Unmarshal(jsonStr3, &result3)
assert.NoError(t, err)
for _, v := range result3 {
kBytes3, _ := hexutil.Decode(v.(map[string]interface{})["content_key"].(string))
vBytes3, _ := hexutil.Decode(v.(map[string]interface{})["content_value"].(string))
contentKey3 := storage.NewContentKey(LightClientBootstrap, kBytes3).Encode()
id = node1.portalProtocol.Self().ID()
_, err = node1.portalProtocol.Gossip(&id, [][]byte{contentKey3}, [][]byte{vBytes3})
assert.NoError(t, err)
time.Sleep(3 * time.Second)
contentId3 := node2.portalProtocol.ToContentId(contentKey3)
get3, err := node2.portalProtocol.Get(contentKey3, contentId3)
assert.NoError(t, err)
assert.Equal(t, vBytes3, get3)
}
}

6
portalnetwork/beacon/portal_api.go
View file

@ -27,14 +27,14 @@ func (api *PortalLightApi) GetCheckpointData(checkpointHash tree.Root) (*capella
}
func (api *PortalLightApi) GetFinalityData() (*capella.LightClientFinalityUpdate, error) {
expectedCurrentSlot := api.bn.Spec.TimeToSlot(zrntcommon.Timestamp(time.Now().Unix()), zrntcommon.Timestamp(BeaconGenesisTime))
recentEpochStart := expectedCurrentSlot - (expectedCurrentSlot % api.bn.Spec.SLOTS_PER_EPOCH) + 1
expectedCurrentSlot := api.bn.spec.TimeToSlot(zrntcommon.Timestamp(time.Now().Unix()), zrntcommon.Timestamp(BeaconGenesisTime))
recentEpochStart := expectedCurrentSlot - (expectedCurrentSlot % api.bn.spec.SLOTS_PER_EPOCH) + 1
return api.bn.GetFinalityUpdate(uint64(recentEpochStart))
}
func (api *PortalLightApi) GetOptimisticData() (*capella.LightClientOptimisticUpdate, error) {
expectedCurrentSlot := api.bn.Spec.TimeToSlot(zrntcommon.Timestamp(time.Now().Unix()), zrntcommon.Timestamp(BeaconGenesisTime))
expectedCurrentSlot := api.bn.spec.TimeToSlot(zrntcommon.Timestamp(time.Now().Unix()), zrntcommon.Timestamp(BeaconGenesisTime))
return api.bn.GetOptimisticUpdate(uint64(expectedCurrentSlot))
}

2
portalnetwork/history/history_network.go
View file

@ -508,7 +508,7 @@ func (h *HistoryNetwork) processContentLoop(ctx context.Context) {
h.log.Error("validate content failed", "err", err)
continue
}
gossippedNum, err := h.portalProtocol.NeighborhoodGossip(&contentElement.Node, contentElement.ContentKeys, contentElement.Contents)
gossippedNum, err := h.portalProtocol.Gossip(&contentElement.Node, contentElement.ContentKeys, contentElement.Contents)
h.log.Trace("gossippedNum", "gossippedNum", gossippedNum)
if err != nil {
h.log.Error("gossip failed", "err", err)

18
portalnetwork/history/history_network_test.go
View file

@ -362,22 +362,6 @@ func parseBlockHeaderKeyContent() ([]contentEntry, error) {
return res, nil
}
type MockStorage struct {
db map[string][]byte
}
func (m *MockStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) {
if content, ok := m.db[string(contentId)]; ok {
return content, nil
}
return nil, storage.ErrContentNotFound
}
func (m *MockStorage) Put(contentKey []byte, contentId []byte, content []byte) error {
m.db[string(contentId)] = content
return nil
}
func genHistoryNetwork(addr string, bootNodes []*enode.Node) (*HistoryNetwork, error) {
glogger := log.NewGlogHandler(log.NewTerminalHandler(os.Stderr, true))
slogVerbosity := log.FromLegacyLevel(5)
@ -448,7 +432,7 @@ func genHistoryNetwork(addr string, bootNodes []*enode.Node) (*HistoryNetwork, e
contentQueue := make(chan *discover.ContentElement, 50)
portalProtocol, err := discover.NewPortalProtocol(conf, string(portalwire.HistoryNetwork), privKey, conn, localNode, discV5, &MockStorage{db: make(map[string][]byte)}, contentQueue)
portalProtocol, err := discover.NewPortalProtocol(conf, string(portalwire.HistoryNetwork), privKey, conn, localNode, discV5, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue)
if err != nil {
return nil, err
}

20
portalnetwork/storage/content_storage.go
View file

@ -11,10 +11,10 @@ type ContentKey struct {
data []byte
}
func NewContentKey(selector ContentType, hash []byte) *ContentKey {
func NewContentKey(selector ContentType, data []byte) *ContentKey {
return &ContentKey{
selector: selector,
data: hash,
data: data,
}
}
@ -30,3 +30,19 @@ type ContentStorage interface {
Put(contentKey []byte, contentId []byte, content []byte) error
}
type MockStorage struct {
Db map[string][]byte
}
func (m *MockStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) {
if content, ok := m.Db[string(contentId)]; ok {
return content, nil
}
return nil, ErrContentNotFound
}
func (m *MockStorage) Put(contentKey []byte, contentId []byte, content []byte) error {
m.Db[string(contentId)] = content
return nil
}