diff --git a/p2p/discover/api.go b/p2p/discover/api.go index 707205eac5..3029946b7e 100644 --- a/p2p/discover/api.go +++ b/p2p/discover/api.go @@ -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) { diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index cf53b1cf6d..6b042de313 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -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 { diff --git a/p2p/discover/portal_protocol_test.go b/p2p/discover/portal_protocol_test.go index 20f7bdb73e..b8b2e728bf 100644 --- a/p2p/discover/portal_protocol_test.go +++ b/p2p/discover/portal_protocol_test.go @@ -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) diff --git a/portalnetwork/beacon/beacon_network.go b/portalnetwork/beacon/beacon_network.go index a906d08f82..f097697f7a 100644 --- a/portalnetwork/beacon/beacon_network.go +++ b/portalnetwork/beacon/beacon_network.go @@ -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 + } + } + } +} diff --git a/portalnetwork/beacon/beacon_network_test.go b/portalnetwork/beacon/beacon_network_test.go new file mode 100644 index 0000000000..9775dd0c8f --- /dev/null +++ b/portalnetwork/beacon/beacon_network_test.go @@ -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) + } +} diff --git a/portalnetwork/beacon/portal_api.go b/portalnetwork/beacon/portal_api.go index afd1b678ae..38948d63e7 100644 --- a/portalnetwork/beacon/portal_api.go +++ b/portalnetwork/beacon/portal_api.go @@ -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)) } diff --git a/portalnetwork/history/history_network.go b/portalnetwork/history/history_network.go index 893a2cb6cc..dc0293a976 100644 --- a/portalnetwork/history/history_network.go +++ b/portalnetwork/history/history_network.go @@ -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) diff --git a/portalnetwork/history/history_network_test.go b/portalnetwork/history/history_network_test.go index 03f59335c6..4040caf677 100644 --- a/portalnetwork/history/history_network_test.go +++ b/portalnetwork/history/history_network_test.go @@ -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 } diff --git a/portalnetwork/storage/content_storage.go b/portalnetwork/storage/content_storage.go index d54334b18a..a894ba70e3 100644 --- a/portalnetwork/storage/content_storage.go +++ b/portalnetwork/storage/content_storage.go @@ -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 +}