From 81aeba27982ed90f148d0854d8328fb3fe1f4359 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Tue, 7 Jan 2020 11:36:05 +0800 Subject: [PATCH] les: move protocol definition to separate package --- ethstats/ethstats.go | 3 +- les/benchmark.go | 30 ++- les/client.go | 9 +- les/client_handler.go | 75 +++---- les/commons.go | 20 +- les/costtracker.go | 146 +++++-------- les/fetcher.go | 7 +- les/handler_test.go | 203 ++++++++--------- les/odr_requests.go | 101 +++------ les/peer.go | 244 +++++++++------------ les/peer_test.go | 111 +++++----- les/protocol.go | 237 -------------------- les/protocol/keyvalueset.go | 86 ++++++++ les/protocol/keyvalueset_test.go | 47 ++++ les/protocol/protocol.go | 362 +++++++++++++++++++++++++++++++ les/retrieve.go | 7 +- les/server.go | 9 +- les/server_handler.go | 135 ++++++------ les/test_helper.go | 49 +++-- les/ulc_test.go | 7 +- 20 files changed, 1021 insertions(+), 867 deletions(-) delete mode 100644 les/protocol.go create mode 100644 les/protocol/keyvalueset.go create mode 100644 les/protocol/keyvalueset_test.go create mode 100644 les/protocol/protocol.go diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go index f9284722cf..07ff5fcac2 100644 --- a/ethstats/ethstats.go +++ b/ethstats/ethstats.go @@ -38,6 +38,7 @@ import ( "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/les" + lesproto "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rpc" @@ -378,7 +379,7 @@ func (s *Service) login(conn *websocket.Conn) error { protocol = fmt.Sprintf("eth/%d", eth.ProtocolVersions[0]) } else { network = fmt.Sprintf("%d", infos.Protocols["les"].(*les.NodeInfo).Network) - protocol = fmt.Sprintf("les/%d", les.ClientProtocolVersions[0]) + protocol = fmt.Sprintf("les/%d", lesproto.ClientProtocolVersions[0]) } auth := &authMsg{ ID: s.node, diff --git a/les/benchmark.go b/les/benchmark.go index 993cf7df15..fec4347df1 100644 --- a/les/benchmark.go +++ b/les/benchmark.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/les/utilities" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" @@ -119,9 +120,9 @@ func (b *benchmarkProofsOrCode) request(peer *peer, index int) error { key := make([]byte, 32) rand.Read(key) if b.code { - return peer.RequestCode(0, 0, []CodeReq{{BHash: b.headHash, AccKey: key}}) + return peer.RequestCode(0, 0, []protocol.CodeRequest{{BlockHash: b.headHash, Account: key}}) } else { - return peer.RequestProofs(0, 0, []ProofReq{{BHash: b.headHash, Key: key}}) + return peer.RequestProofs(0, 0, []protocol.TrieProofRequest{{BlockHash: b.headHash, Key: key}}) } } @@ -146,7 +147,7 @@ func (b *benchmarkHelperTrie) init(h *serverHandler, count int) error { } func (b *benchmarkHelperTrie) request(peer *peer, index int) error { - reqs := make([]HelperTrieReq, b.reqCount) + reqs := make([]protocol.HelperTrieRequest, b.reqCount) if b.bloom { bitIdx := uint16(rand.Intn(2048)) @@ -154,13 +155,22 @@ func (b *benchmarkHelperTrie) request(peer *peer, index int) error { key := make([]byte, 10) binary.BigEndian.PutUint16(key[:2], bitIdx) binary.BigEndian.PutUint64(key[2:], uint64(rand.Int63n(int64(b.sectionCount)))) - reqs[i] = HelperTrieReq{Type: htBloomBits, TrieIdx: b.sectionCount - 1, Key: key} + reqs[i] = protocol.HelperTrieRequest{ + Type: protocol.HelperTrieBloomTrie, + TrieIndex: b.sectionCount - 1, + Key: key, + } } } else { for i := range reqs { key := make([]byte, 8) binary.BigEndian.PutUint64(key[:], uint64(rand.Int63n(int64(b.headNum)))) - reqs[i] = HelperTrieReq{Type: htCanonical, TrieIdx: b.sectionCount - 1, Key: key, AuxReq: auxHeader} + reqs[i] = protocol.HelperTrieRequest{ + Type: protocol.HelperTrieCHT, + TrieIndex: b.sectionCount - 1, + Key: key, + AuxType: protocol.AuxHeader, + } } } @@ -284,13 +294,13 @@ func (h *serverHandler) measure(setup *benchmarkSetup, count int) error { var id enode.ID rand.Read(id[:]) - clientPeer := newPeer(lpv2, NetworkId, false, p2p.NewPeer(id, "client", nil), clientMeteredPipe) - serverPeer := newPeer(lpv2, NetworkId, false, p2p.NewPeer(id, "server", nil), serverMeteredPipe) + clientPeer := newPeer(protocol.Lpv2, protocol.NetworkId, false, p2p.NewPeer(id, "client", nil), clientMeteredPipe) + serverPeer := newPeer(protocol.Lpv2, protocol.NetworkId, false, p2p.NewPeer(id, "server", nil), serverMeteredPipe) serverPeer.sendQueue = utilities.NewExecQueue(count) serverPeer.announceType = announceTypeNone - serverPeer.fcCosts = make(requestCostTable) - c := &requestCosts{} - for code := range requests { + serverPeer.fcCosts = make(protocol.RequestCostTable) + c := &protocol.RequestCost{} + for code := range protocol.LesRequests { serverPeer.fcCosts[code] = c } serverPeer.fcParams = flowcontrol.ServerParams{BufLimit: 1, MinRecharge: 1} diff --git a/les/client.go b/les/client.go index 579062e7fd..8d10810f93 100644 --- a/les/client.go +++ b/les/client.go @@ -37,6 +37,7 @@ import ( "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/internal/ethapi" "github.com/ethereum/go-ethereum/les/checkpointoracle" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" @@ -217,14 +218,14 @@ func (s *LightEthereum) ResetWithGenesisBlock(gb *types.Block) { func (s *LightEthereum) BlockChain() *light.LightChain { return s.blockchain } func (s *LightEthereum) TxPool() *light.TxPool { return s.txPool } func (s *LightEthereum) Engine() consensus.Engine { return s.engine } -func (s *LightEthereum) LesVersion() int { return int(ClientProtocolVersions[0]) } +func (s *LightEthereum) LesVersion() int { return int(protocol.ClientProtocolVersions[0]) } func (s *LightEthereum) Downloader() *downloader.Downloader { return s.handler.downloader } func (s *LightEthereum) EventMux() *event.TypeMux { return s.eventMux } // Protocols implements node.Service, returning all the currently configured // network protocols to start. func (s *LightEthereum) Protocols() []p2p.Protocol { - return s.makeProtocols(ClientProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} { + return s.makeProtocols(protocol.ClientProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} { if p := s.peers.Peer(peerIdToString(id)); p != nil { return p.Info() } @@ -244,8 +245,8 @@ func (s *LightEthereum) Start(srvr *p2p.Server) error { s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.config.NetworkId) // clients are searching for the first advertised protocol in the list - protocolVersion := AdvertiseProtocolVersions[0] - s.serverPool.start(srvr, lesTopic(s.blockchain.Genesis().Hash(), protocolVersion)) + protocolVersion := protocol.AdvertiseProtocolVersions[0] + s.serverPool.start(srvr, protocol.LesTopic(s.blockchain.Genesis().Hash(), protocolVersion)) return nil } diff --git a/les/client_handler.go b/les/client_handler.go index 7fdb165719..e26d6abe20 100644 --- a/les/client_handler.go +++ b/les/client_handler.go @@ -25,6 +25,7 @@ import ( "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth/downloader" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" @@ -151,8 +152,8 @@ func (h *clientHandler) handleMsg(p *peer) error { } p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size) - if msg.Size > ProtocolMaxMsgSize { - return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + if msg.Size > protocol.ProtocolMaxMsgSize { + return protocol.ErrResp(protocol.ErrMsgTooLarge, "%v > %v", msg.Size, protocol.ProtocolMaxMsgSize) } defer msg.Discard() @@ -160,43 +161,43 @@ func (h *clientHandler) handleMsg(p *peer) error { // Handle the message depending on its contents switch msg.Code { - case AnnounceMsg: + case protocol.AnnounceMsg: p.Log().Trace("Received announce message") - var req announceData - if err := msg.Decode(&req); err != nil { - return errResp(ErrDecode, "%v: %v", msg, err) + var anno protocol.Announcement + if err := msg.Decode(&anno); err != nil { + return protocol.ErrResp(protocol.ErrDecode, "%v: %v", msg, err) } - if err := req.sanityCheck(); err != nil { + if err := anno.SanityCheck(); err != nil { return err } - update, size := req.Update.decode() + update, size := anno.Update.ToMap() if p.rejectUpdate(size) { - return errResp(ErrRequestRejected, "") + return protocol.ErrResp(protocol.ErrRequestRejected, "") } p.updateFlowControl(update) - if req.Hash != (common.Hash{}) { + if anno.Hash != (common.Hash{}) { if p.announceType == announceTypeNone { - return errResp(ErrUnexpectedResponse, "") + return protocol.ErrResp(protocol.ErrUnexpectedResponse, "") } if p.announceType == announceTypeSigned { - if err := req.checkSignature(p.ID(), update); err != nil { + if err := anno.CheckSignature(p.ID(), update); err != nil { p.Log().Trace("Invalid announcement signature", "err", err) return err } p.Log().Trace("Valid announcement signature") } - p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth) - h.fetcher.announce(p, &req) + p.Log().Trace("Announce message content", "number", anno.Number, "hash", anno.Hash, "td", anno.Td, "reorg", anno.ReorgDepth) + h.fetcher.announce(p, &anno) } - case BlockHeadersMsg: + case protocol.BlockHeadersMsg: p.Log().Trace("Received block header response message") var resp struct { ReqID, BV uint64 Headers []*types.Header } if err := msg.Decode(&resp); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ReceivedReply(resp.ReqID, resp.BV) if h.fetcher.requestedID(resp.ReqID) { @@ -206,14 +207,14 @@ func (h *clientHandler) handleMsg(p *peer) error { log.Debug("Failed to deliver headers", "err", err) } } - case BlockBodiesMsg: + case protocol.BlockBodiesMsg: p.Log().Trace("Received block bodies response") var resp struct { ReqID, BV uint64 Data []*types.Body } if err := msg.Decode(&resp); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ @@ -221,14 +222,14 @@ func (h *clientHandler) handleMsg(p *peer) error { ReqID: resp.ReqID, Obj: resp.Data, } - case CodeMsg: + case protocol.CodeMsg: p.Log().Trace("Received code response") var resp struct { ReqID, BV uint64 Data [][]byte } if err := msg.Decode(&resp); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ @@ -236,14 +237,14 @@ func (h *clientHandler) handleMsg(p *peer) error { ReqID: resp.ReqID, Obj: resp.Data, } - case ReceiptsMsg: + case protocol.ReceiptsMsg: p.Log().Trace("Received receipts response") var resp struct { ReqID, BV uint64 Receipts []types.Receipts } if err := msg.Decode(&resp); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ @@ -251,14 +252,14 @@ func (h *clientHandler) handleMsg(p *peer) error { ReqID: resp.ReqID, Obj: resp.Receipts, } - case ProofsV2Msg: + case protocol.ProofsV2Msg: p.Log().Trace("Received les/2 proofs response") var resp struct { ReqID, BV uint64 Data light.NodeList } if err := msg.Decode(&resp); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ @@ -266,14 +267,14 @@ func (h *clientHandler) handleMsg(p *peer) error { ReqID: resp.ReqID, Obj: resp.Data, } - case HelperTrieProofsMsg: + case protocol.HelperTrieProofsMsg: p.Log().Trace("Received helper trie proof response") var resp struct { ReqID, BV uint64 - Data HelperTrieResps + Data protocol.HelperTrieResponse } if err := msg.Decode(&resp); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ @@ -281,14 +282,14 @@ func (h *clientHandler) handleMsg(p *peer) error { ReqID: resp.ReqID, Obj: resp.Data, } - case TxStatusMsg: + case protocol.TxStatusMsg: p.Log().Trace("Received tx status response") var resp struct { ReqID, BV uint64 Status []light.TxStatus } if err := msg.Decode(&resp); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ @@ -296,21 +297,21 @@ func (h *clientHandler) handleMsg(p *peer) error { ReqID: resp.ReqID, Obj: resp.Status, } - case StopMsg: + case protocol.StopMsg: p.freezeServer(true) h.backend.retriever.frozen(p) p.Log().Debug("Service stopped") - case ResumeMsg: + case protocol.ResumeMsg: var bv uint64 if err := msg.Decode(&bv); err != nil { - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } p.fcServer.ResumeFreeze(bv) p.freezeServer(false) p.Log().Debug("Service resumed") default: p.Log().Trace("Received invalid message", "code", msg.Code) - return errResp(ErrInvalidMsgCode, "%v", msg.Code) + return protocol.ErrResp(protocol.ErrInvalidMsgCode, "%v", msg.Code) } // Deliver the received response to retriever. if deliverMsg != nil { @@ -341,7 +342,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s rq := &distReq{ getCost: func(dp distPeer) uint64 { peer := dp.(*peer) - return peer.GetRequestCost(GetBlockHeadersMsg, amount) + return peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount) }, canSend: func(dp distPeer) bool { return dp.(*peer) == pc.peer @@ -349,7 +350,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s request: func(dp distPeer) func() { reqID := genReqID() peer := dp.(*peer) - cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) + cost := peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount) peer.fcServer.QueuedRequest(reqID, cost) return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) } }, @@ -365,7 +366,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip rq := &distReq{ getCost: func(dp distPeer) uint64 { peer := dp.(*peer) - return peer.GetRequestCost(GetBlockHeadersMsg, amount) + return peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount) }, canSend: func(dp distPeer) bool { return dp.(*peer) == pc.peer @@ -373,7 +374,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip request: func(dp distPeer) func() { reqID := genReqID() peer := dp.(*peer) - cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) + cost := peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount) peer.fcServer.QueuedRequest(reqID, cost) return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) } }, diff --git a/les/commons.go b/les/commons.go index b402c51769..3f8854b2b1 100644 --- a/les/commons.go +++ b/les/commons.go @@ -17,7 +17,6 @@ package les import ( - "fmt" "math/big" "sync" @@ -28,28 +27,13 @@ import ( "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/les/checkpointoracle" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/discv5" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/params" ) -func errResp(code errCode, format string, v ...interface{}) error { - return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...)) -} - -func lesTopic(genesisHash common.Hash, protocolVersion uint) discv5.Topic { - var name string - switch protocolVersion { - case lpv2: - name = "LES2" - default: - panic(nil) - } - return discv5.Topic(name + "@" + common.Bytes2Hex(genesisHash.Bytes()[0:8])) -} - type chainReader interface { CurrentHeader() *types.Header } @@ -89,7 +73,7 @@ func (c *lesCommons) makeProtocols(versions []uint, runPeer func(version uint, p protos[i] = p2p.Protocol{ Name: "les", Version: version, - Length: ProtocolLengths[version], + Length: protocol.ProtocolLengths[version], NodeInfo: c.nodeInfo, Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error { return runPeer(version, peer, rw) diff --git a/les/costtracker.go b/les/costtracker.go index 81da045660..8a25e9586a 100644 --- a/les/costtracker.go +++ b/les/costtracker.go @@ -27,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" ) @@ -35,48 +36,48 @@ const makeCostStats = false // make request cost statistics during operation var ( // average request cost estimates based on serving time - reqAvgTimeCost = requestCostTable{ - GetBlockHeadersMsg: {150000, 30000}, - GetBlockBodiesMsg: {0, 700000}, - GetReceiptsMsg: {0, 1000000}, - GetCodeMsg: {0, 450000}, - GetProofsV2Msg: {0, 600000}, - GetHelperTrieProofsMsg: {0, 1000000}, - SendTxV2Msg: {0, 450000}, - GetTxStatusMsg: {0, 250000}, + reqAvgTimeCost = protocol.RequestCostTable{ + protocol.GetBlockHeadersMsg: {150000, 30000}, + protocol.GetBlockBodiesMsg: {0, 700000}, + protocol.GetReceiptsMsg: {0, 1000000}, + protocol.GetCodeMsg: {0, 450000}, + protocol.GetProofsV2Msg: {0, 600000}, + protocol.GetHelperTrieProofsMsg: {0, 1000000}, + protocol.SendTxV2Msg: {0, 450000}, + protocol.GetTxStatusMsg: {0, 250000}, } // maximum incoming message size estimates - reqMaxInSize = requestCostTable{ - GetBlockHeadersMsg: {40, 0}, - GetBlockBodiesMsg: {0, 40}, - GetReceiptsMsg: {0, 40}, - GetCodeMsg: {0, 80}, - GetProofsV2Msg: {0, 80}, - GetHelperTrieProofsMsg: {0, 20}, - SendTxV2Msg: {0, 16500}, - GetTxStatusMsg: {0, 50}, + reqMaxInSize = protocol.RequestCostTable{ + protocol.GetBlockHeadersMsg: {40, 0}, + protocol.GetBlockBodiesMsg: {0, 40}, + protocol.GetReceiptsMsg: {0, 40}, + protocol.GetCodeMsg: {0, 80}, + protocol.GetProofsV2Msg: {0, 80}, + protocol.GetHelperTrieProofsMsg: {0, 20}, + protocol.SendTxV2Msg: {0, 16500}, + protocol.GetTxStatusMsg: {0, 50}, } // maximum outgoing message size estimates - reqMaxOutSize = requestCostTable{ - GetBlockHeadersMsg: {0, 556}, - GetBlockBodiesMsg: {0, 100000}, - GetReceiptsMsg: {0, 200000}, - GetCodeMsg: {0, 50000}, - GetProofsV2Msg: {0, 4000}, - GetHelperTrieProofsMsg: {0, 4000}, - SendTxV2Msg: {0, 100}, - GetTxStatusMsg: {0, 100}, + reqMaxOutSize = protocol.RequestCostTable{ + protocol.GetBlockHeadersMsg: {0, 556}, + protocol.GetBlockBodiesMsg: {0, 100000}, + protocol.GetReceiptsMsg: {0, 200000}, + protocol.GetCodeMsg: {0, 50000}, + protocol.GetProofsV2Msg: {0, 4000}, + protocol.GetHelperTrieProofsMsg: {0, 4000}, + protocol.SendTxV2Msg: {0, 100}, + protocol.GetTxStatusMsg: {0, 100}, } // request amounts that have to fit into the minimum buffer size minBufferMultiplier times minBufferReqAmount = map[uint64]uint64{ - GetBlockHeadersMsg: 192, - GetBlockBodiesMsg: 1, - GetReceiptsMsg: 1, - GetCodeMsg: 1, - GetProofsV2Msg: 1, - GetHelperTrieProofsMsg: 16, - SendTxV2Msg: 8, - GetTxStatusMsg: 64, + protocol.GetBlockHeadersMsg: 192, + protocol.GetBlockBodiesMsg: 1, + protocol.GetReceiptsMsg: 1, + protocol.GetCodeMsg: 1, + protocol.GetProofsV2Msg: 1, + protocol.GetHelperTrieProofsMsg: 16, + protocol.SendTxV2Msg: 8, + protocol.GetTxStatusMsg: 64, } minBufferMultiplier = 3 ) @@ -131,8 +132,8 @@ type costTracker struct { stats map[uint64][]uint64 // Used for testing purpose. // TestHooks - testing bool // Disable real cost evaluation for testing purpose. - testCostList RequestCostList // Customized cost table for testing purpose. + testing bool // Disable real cost evaluation for testing purpose. + testCostList protocol.RequestCostList // Customized cost table for testing purpose. } // newCostTracker creates a cost tracker and loads the cost factor statistics from the database. @@ -182,7 +183,7 @@ func (ct *costTracker) stop() { // makeCostList returns upper cost estimates based on the hardcoded cost estimate // tables and the optionally specified incoming/outgoing bandwidth limits -func (ct *costTracker) makeCostList(globalFactor float64) RequestCostList { +func (ct *costTracker) makeCostList(globalFactor float64) protocol.RequestCostList { maxCost := func(avgTimeCost, inSize, outSize uint64) uint64 { cost := avgTimeCost * maxCostFactor inSizeCost := uint64(float64(inSize) * ct.inSizeFactor * globalFactor) @@ -195,10 +196,10 @@ func (ct *costTracker) makeCostList(globalFactor float64) RequestCostList { } return cost } - var list RequestCostList + var list protocol.RequestCostList for code, data := range reqAvgTimeCost { - baseCost := maxCost(data.baseCost, reqMaxInSize[code].baseCost, reqMaxOutSize[code].baseCost) - reqCost := maxCost(data.reqCost, reqMaxInSize[code].reqCost, reqMaxOutSize[code].reqCost) + baseCost := maxCost(data.BaseCost, reqMaxInSize[code].BaseCost, reqMaxOutSize[code].BaseCost) + reqCost := maxCost(data.ReqCost, reqMaxInSize[code].ReqCost, reqMaxOutSize[code].ReqCost) if ct.minBufLimit != 0 { // if minBufLimit is set then always enforce maximum request cost <= minBufLimit maxCost := baseCost + reqCost*minBufferReqAmount[code] @@ -209,7 +210,7 @@ func (ct *costTracker) makeCostList(globalFactor float64) RequestCostList { } } - list = append(list, requestCostListItem{ + list = append(list, protocol.RequestCostListItem{ MsgCode: code, BaseCost: baseCost, ReqCost: reqCost, @@ -278,21 +279,21 @@ func (ct *costTracker) gfLoop() { // Record more metrics if we are debugging if metrics.EnabledExpensive { switch r.msgCode { - case GetBlockHeadersMsg: + case protocol.GetBlockHeadersMsg: relativeCostHeaderHistogram.Update(relCost) - case GetBlockBodiesMsg: + case protocol.GetBlockBodiesMsg: relativeCostBodyHistogram.Update(relCost) - case GetReceiptsMsg: + case protocol.GetReceiptsMsg: relativeCostReceiptHistogram.Update(relCost) - case GetCodeMsg: + case protocol.GetCodeMsg: relativeCostCodeHistogram.Update(relCost) - case GetProofsV2Msg: + case protocol.GetProofsV2Msg: relativeCostProofHistogram.Update(relCost) - case GetHelperTrieProofsMsg: + case protocol.GetHelperTrieProofsMsg: relativeCostHelperProofHistogram.Update(relCost) - case SendTxV2Msg: + case protocol.SendTxV2Msg: relativeCostSendTxHistogram.Update(relCost) - case GetTxStatusMsg: + case protocol.GetTxStatusMsg: relativeCostTxStatusHistogram.Update(relCost) } } @@ -302,7 +303,7 @@ func (ct *costTracker) gfLoop() { // requests involve txpool query, which is usually unstable. // // TODO(rjl493456442) fixes this. - if r.msgCode == SendTxV2Msg || r.msgCode == GetTxStatusMsg { + if r.msgCode == protocol.SendTxV2Msg || r.msgCode == protocol.GetTxStatusMsg { continue } requestServedMeter.Mark(int64(r.servingTime)) @@ -410,7 +411,7 @@ func (ct *costTracker) subscribeTotalRecharge(ch chan uint64) uint64 { // average estimate statistics func (ct *costTracker) updateStats(code, amount, servingTime, realCost uint64) { avg := reqAvgTimeCost[code] - avgTimeCost := avg.baseCost + amount*avg.reqCost + avgTimeCost := avg.BaseCost + amount*avg.ReqCost select { case ct.reqInfoCh <- reqInfo{float64(avgTimeCost), float64(servingTime), code}: default: @@ -457,46 +458,9 @@ func (ct *costTracker) printStats() { } } -type ( - // requestCostTable assigns a cost estimate function to each request type - // which is a linear function of the requested amount - // (cost = baseCost + reqCost * amount) - requestCostTable map[uint64]*requestCosts - requestCosts struct { - baseCost, reqCost uint64 - } - - // RequestCostList is a list representation of request costs which is used for - // database storage and communication through the network - RequestCostList []requestCostListItem - requestCostListItem struct { - MsgCode, BaseCost, ReqCost uint64 - } -) - -// getMaxCost calculates the estimated cost for a given request type and amount -func (table requestCostTable) getMaxCost(code, amount uint64) uint64 { - costs := table[code] - return costs.baseCost + amount*costs.reqCost -} - -// decode converts a cost list to a cost table -func (list RequestCostList) decode(protocolLength uint64) requestCostTable { - table := make(requestCostTable) - for _, e := range list { - if e.MsgCode < protocolLength { - table[e.MsgCode] = &requestCosts{ - baseCost: e.BaseCost, - reqCost: e.ReqCost, - } - } - } - return table -} - // testCostList returns a dummy request cost list used by tests -func testCostList(testCost uint64) RequestCostList { - cl := make(RequestCostList, len(reqAvgTimeCost)) +func testCostList(testCost uint64) protocol.RequestCostList { + cl := make(protocol.RequestCostList, len(reqAvgTimeCost)) var max uint64 for code := range reqAvgTimeCost { if code > max { diff --git a/les/fetcher.go b/les/fetcher.go index df76c56d70..45536074d5 100644 --- a/les/fetcher.go +++ b/les/fetcher.go @@ -26,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" ) @@ -250,7 +251,7 @@ func (f *lightFetcher) unregisterPeer(p *peer) { // announce processes a new announcement message received from a peer, adding new // nodes to the peer's block tree and removing old nodes if necessary -func (f *lightFetcher) announce(p *peer, head *announceData) { +func (f *lightFetcher) announce(p *peer, head *protocol.Announcement) { f.lock.Lock() defer f.lock.Unlock() p.Log().Debug("Received new announcement", "number", head.Number, "hash", head.Hash, "reorg", head.ReorgDepth) @@ -519,7 +520,7 @@ func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bes return &distReq{ getCost: func(dp distPeer) uint64 { p := dp.(*peer) - return p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount)) + return p.GetRequestCost(protocol.GetBlockHeadersMsg, int(bestAmount)) }, canSend: func(dp distPeer) bool { p := dp.(*peer) @@ -548,7 +549,7 @@ func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bes } f.lock.Unlock() - cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount)) + cost := p.GetRequestCost(protocol.GetBlockHeadersMsg, int(bestAmount)) p.fcServer.QueuedRequest(reqID, cost) f.reqMu.Lock() f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()} diff --git a/les/handler_test.go b/les/handler_test.go index aad8d18e45..e07c27b5a4 100644 --- a/les/handler_test.go +++ b/les/handler_test.go @@ -31,6 +31,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth/downloader" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/params" @@ -50,8 +51,8 @@ func expectResponse(r p2p.MsgReader, msgcode, reqID, bv uint64, data interface{} func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) } func TestGetBlockHeadersLes3(t *testing.T) { testGetBlockHeaders(t, 3) } -func testGetBlockHeaders(t *testing.T, protocol int) { - server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil, false, true, 0) +func testGetBlockHeaders(t *testing.T, p int) { + server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, p, nil, false, true, 0) defer tearDown() bc := server.handler.blockchain @@ -62,29 +63,29 @@ func testGetBlockHeaders(t *testing.T, protocol int) { unknown[i] = byte(i) } // Create a batch of tests for various scenarios - limit := uint64(MaxHeaderFetch) + limit := uint64(protocol.MaxHeaderFetch) tests := []struct { - query *getBlockHeadersData // The query to execute for header retrieval - expect []common.Hash // The hashes of the block whose headers are expected + query *protocol.GetBlockHeadersRequest // The query to execute for header retrieval + expect []common.Hash // The hashes of the block whose headers are expected }{ // A single random block should be retrievable by hash and number too { - &getBlockHeadersData{Origin: hashOrNumber{Hash: bc.GetBlockByNumber(limit / 2).Hash()}, Amount: 1}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: bc.GetBlockByNumber(limit / 2).Hash()}, Amount: 1}, []common.Hash{bc.GetBlockByNumber(limit / 2).Hash()}, }, { - &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 1}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Amount: 1}, []common.Hash{bc.GetBlockByNumber(limit / 2).Hash()}, }, // Multiple headers should be retrievable in both directions { - &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Amount: 3}, []common.Hash{ bc.GetBlockByNumber(limit / 2).Hash(), bc.GetBlockByNumber(limit/2 + 1).Hash(), bc.GetBlockByNumber(limit/2 + 2).Hash(), }, }, { - &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3, Reverse: true}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Amount: 3, Reverse: true}, []common.Hash{ bc.GetBlockByNumber(limit / 2).Hash(), bc.GetBlockByNumber(limit/2 - 1).Hash(), @@ -93,14 +94,14 @@ func testGetBlockHeaders(t *testing.T, protocol int) { }, // Multiple headers with skip lists should be retrievable { - &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3}, []common.Hash{ bc.GetBlockByNumber(limit / 2).Hash(), bc.GetBlockByNumber(limit/2 + 4).Hash(), bc.GetBlockByNumber(limit/2 + 8).Hash(), }, }, { - &getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3, Reverse: true}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3, Reverse: true}, []common.Hash{ bc.GetBlockByNumber(limit / 2).Hash(), bc.GetBlockByNumber(limit/2 - 4).Hash(), @@ -109,26 +110,26 @@ func testGetBlockHeaders(t *testing.T, protocol int) { }, // The chain endpoints should be retrievable { - &getBlockHeadersData{Origin: hashOrNumber{Number: 0}, Amount: 1}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: 0}, Amount: 1}, []common.Hash{bc.GetBlockByNumber(0).Hash()}, }, { - &getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64()}, Amount: 1}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64()}, Amount: 1}, []common.Hash{bc.CurrentBlock().Hash()}, }, - // Ensure protocol limits are honored + // Ensure p limits are honored //{ - // &getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() - 1}, Amount: limit + 10, Reverse: true}, + // &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() - 1}, Amount: limit + 10, Reverse: true}, // []common.Hash{}, //}, // Check that requesting more than available is handled gracefully { - &getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 3, Amount: 3}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 3, Amount: 3}, []common.Hash{ bc.GetBlockByNumber(bc.CurrentBlock().NumberU64() - 4).Hash(), bc.GetBlockByNumber(bc.CurrentBlock().NumberU64()).Hash(), }, }, { - &getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 3, Amount: 3, Reverse: true}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: 4}, Skip: 3, Amount: 3, Reverse: true}, []common.Hash{ bc.GetBlockByNumber(4).Hash(), bc.GetBlockByNumber(0).Hash(), @@ -136,13 +137,13 @@ func testGetBlockHeaders(t *testing.T, protocol int) { }, // Check that requesting more than available is handled gracefully, even if mid skip { - &getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 2, Amount: 3}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 2, Amount: 3}, []common.Hash{ bc.GetBlockByNumber(bc.CurrentBlock().NumberU64() - 4).Hash(), bc.GetBlockByNumber(bc.CurrentBlock().NumberU64() - 1).Hash(), }, }, { - &getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 2, Amount: 3, Reverse: true}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: 4}, Skip: 2, Amount: 3, Reverse: true}, []common.Hash{ bc.GetBlockByNumber(4).Hash(), bc.GetBlockByNumber(1).Hash(), @@ -150,10 +151,10 @@ func testGetBlockHeaders(t *testing.T, protocol int) { }, // Check that non existing headers aren't returned { - &getBlockHeadersData{Origin: hashOrNumber{Hash: unknown}, Amount: 1}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: unknown}, Amount: 1}, []common.Hash{}, }, { - &getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() + 1}, Amount: 1}, + &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() + 1}, Amount: 1}, []common.Hash{}, }, } @@ -168,9 +169,9 @@ func testGetBlockHeaders(t *testing.T, protocol int) { // Send the hash request and verify the response reqID++ - cost := server.peer.peer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount)) - sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, cost, tt.query) - if err := expectResponse(server.peer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetBlockHeadersMsg, int(tt.query.Amount)) + sendRequest(server.peer.app, protocol.GetBlockHeadersMsg, reqID, cost, tt.query) + if err := expectResponse(server.peer.app, protocol.BlockHeadersMsg, reqID, testBufLimit, headers); err != nil { t.Errorf("test %d: headers mismatch: %v", i, err) } } @@ -180,14 +181,14 @@ func testGetBlockHeaders(t *testing.T, protocol int) { func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) } func TestGetBlockBodiesLes3(t *testing.T) { testGetBlockBodies(t, 3) } -func testGetBlockBodies(t *testing.T, protocol int) { - server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, nil, false, true, 0) +func testGetBlockBodies(t *testing.T, p int) { + server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, p, nil, false, true, 0) defer tearDown() bc := server.handler.blockchain // Create a batch of tests for various scenarios - limit := MaxBodyFetch + limit := protocol.MaxBodyFetch tests := []struct { random int // Number of blocks to fetch randomly from the chain explicit []common.Hash // Explicitly requested blocks @@ -246,9 +247,9 @@ func testGetBlockBodies(t *testing.T, protocol int) { reqID++ // Send the hash request and verify the response - cost := server.peer.peer.GetRequestCost(GetBlockBodiesMsg, len(hashes)) - sendRequest(server.peer.app, GetBlockBodiesMsg, reqID, cost, hashes) - if err := expectResponse(server.peer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetBlockBodiesMsg, len(hashes)) + sendRequest(server.peer.app, protocol.GetBlockBodiesMsg, reqID, cost, hashes) + if err := expectResponse(server.peer.app, protocol.BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil { t.Errorf("test %d: bodies mismatch: %v", i, err) } } @@ -258,19 +259,19 @@ func testGetBlockBodies(t *testing.T, protocol int) { func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) } func TestGetCodeLes3(t *testing.T) { testGetCode(t, 3) } -func testGetCode(t *testing.T, protocol int) { +func testGetCode(t *testing.T, p int) { // Assemble the test environment - server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0) + server, tearDown := newServerEnv(t, 4, p, nil, false, true, 0) defer tearDown() bc := server.handler.blockchain - var codereqs []*CodeReq + var codereqs []protocol.CodeRequest var codes [][]byte for i := uint64(0); i <= bc.CurrentBlock().NumberU64(); i++ { header := bc.GetHeaderByNumber(i) - req := &CodeReq{ - BHash: header.Hash(), - AccKey: crypto.Keccak256(testContractAddr[:]), + req := protocol.CodeRequest{ + BlockHash: header.Hash(), + Account: crypto.Keccak256(testContractAddr[:]), } codereqs = append(codereqs, req) if i >= testContractDeployed { @@ -278,9 +279,9 @@ func testGetCode(t *testing.T, protocol int) { } } - cost := server.peer.peer.GetRequestCost(GetCodeMsg, len(codereqs)) - sendRequest(server.peer.app, GetCodeMsg, 42, cost, codereqs) - if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, codes); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetCodeMsg, len(codereqs)) + sendRequest(server.peer.app, protocol.GetCodeMsg, 42, cost, codereqs) + if err := expectResponse(server.peer.app, protocol.CodeMsg, 42, testBufLimit, codes); err != nil { t.Errorf("codes mismatch: %v", err) } } @@ -289,19 +290,19 @@ func testGetCode(t *testing.T, protocol int) { func TestGetStaleCodeLes2(t *testing.T) { testGetStaleCode(t, 2) } func TestGetStaleCodeLes3(t *testing.T) { testGetStaleCode(t, 3) } -func testGetStaleCode(t *testing.T, protocol int) { - server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0) +func testGetStaleCode(t *testing.T, p int) { + server, tearDown := newServerEnv(t, core.TriesInMemory+4, p, nil, false, true, 0) defer tearDown() bc := server.handler.blockchain check := func(number uint64, expected [][]byte) { - req := &CodeReq{ - BHash: bc.GetHeaderByNumber(number).Hash(), - AccKey: crypto.Keccak256(testContractAddr[:]), + req := protocol.CodeRequest{ + BlockHash: bc.GetHeaderByNumber(number).Hash(), + Account: crypto.Keccak256(testContractAddr[:]), } - cost := server.peer.peer.GetRequestCost(GetCodeMsg, 1) - sendRequest(server.peer.app, GetCodeMsg, 42, cost, []*CodeReq{req}) - if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, expected); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetCodeMsg, 1) + sendRequest(server.peer.app, protocol.GetCodeMsg, 42, cost, []protocol.CodeRequest{req}) + if err := expectResponse(server.peer.app, protocol.CodeMsg, 42, testBufLimit, expected); err != nil { t.Errorf("codes mismatch: %v", err) } } @@ -314,9 +315,9 @@ func testGetStaleCode(t *testing.T, protocol int) { func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) } func TestGetReceiptLes3(t *testing.T) { testGetReceipt(t, 3) } -func testGetReceipt(t *testing.T, protocol int) { +func testGetReceipt(t *testing.T, p int) { // Assemble the test environment - server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0) + server, tearDown := newServerEnv(t, 4, p, nil, false, true, 0) defer tearDown() bc := server.handler.blockchain @@ -331,9 +332,9 @@ func testGetReceipt(t *testing.T, protocol int) { receipts = append(receipts, rawdb.ReadRawReceipts(server.db, block.Hash(), block.NumberU64())) } // Send the hash request and verify the response - cost := server.peer.peer.GetRequestCost(GetReceiptsMsg, len(hashes)) - sendRequest(server.peer.app, GetReceiptsMsg, 42, cost, hashes) - if err := expectResponse(server.peer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetReceiptsMsg, len(hashes)) + sendRequest(server.peer.app, protocol.GetReceiptsMsg, 42, cost, hashes) + if err := expectResponse(server.peer.app, protocol.ReceiptsMsg, 42, testBufLimit, receipts); err != nil { t.Errorf("receipts mismatch: %v", err) } } @@ -342,14 +343,14 @@ func testGetReceipt(t *testing.T, protocol int) { func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) } func TestGetProofsLes3(t *testing.T) { testGetProofs(t, 3) } -func testGetProofs(t *testing.T, protocol int) { +func testGetProofs(t *testing.T, p int) { // Assemble the test environment - server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0) + server, tearDown := newServerEnv(t, 4, p, nil, false, true, 0) defer tearDown() bc := server.handler.blockchain - var proofreqs []ProofReq + var proofreqs []protocol.TrieProofRequest proofsV2 := light.NewNodeSet() accounts := []common.Address{bankAddr, userAddr1, userAddr2, signerAddr, {}} @@ -358,18 +359,18 @@ func testGetProofs(t *testing.T, protocol int) { trie, _ := trie.New(header.Root, trie.NewDatabase(server.db)) for _, acc := range accounts { - req := ProofReq{ - BHash: header.Hash(), - Key: crypto.Keccak256(acc[:]), + req := protocol.TrieProofRequest{ + BlockHash: header.Hash(), + Key: crypto.Keccak256(acc[:]), } proofreqs = append(proofreqs, req) trie.Prove(crypto.Keccak256(acc[:]), 0, proofsV2) } } // Send the proof request and verify the response - cost := server.peer.peer.GetRequestCost(GetProofsV2Msg, len(proofreqs)) - sendRequest(server.peer.app, GetProofsV2Msg, 42, cost, proofreqs) - if err := expectResponse(server.peer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetProofsV2Msg, len(proofreqs)) + sendRequest(server.peer.app, protocol.GetProofsV2Msg, 42, cost, proofreqs) + if err := expectResponse(server.peer.app, protocol.ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil { t.Errorf("proofs mismatch: %v", err) } } @@ -378,8 +379,8 @@ func testGetProofs(t *testing.T, protocol int) { func TestGetStaleProofLes2(t *testing.T) { testGetStaleProof(t, 2) } func TestGetStaleProofLes3(t *testing.T) { testGetStaleProof(t, 3) } -func testGetStaleProof(t *testing.T, protocol int) { - server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0) +func testGetStaleProof(t *testing.T, p int) { + server, tearDown := newServerEnv(t, core.TriesInMemory+4, p, nil, false, true, 0) defer tearDown() bc := server.handler.blockchain @@ -388,12 +389,12 @@ func testGetStaleProof(t *testing.T, protocol int) { header = bc.GetHeaderByNumber(number) account = crypto.Keccak256(userAddr1.Bytes()) ) - req := &ProofReq{ - BHash: header.Hash(), - Key: account, + req := &protocol.TrieProofRequest{ + BlockHash: header.Hash(), + Key: account, } - cost := server.peer.peer.GetRequestCost(GetProofsV2Msg, 1) - sendRequest(server.peer.app, GetProofsV2Msg, 42, cost, []*ProofReq{req}) + cost := server.peer.peer.GetRequestCost(protocol.GetProofsV2Msg, 1) + sendRequest(server.peer.app, protocol.GetProofsV2Msg, 42, cost, []*protocol.TrieProofRequest{req}) var expected []rlp.RawValue if wantOK { @@ -402,7 +403,7 @@ func testGetStaleProof(t *testing.T, protocol int) { t.Prove(account, 0, proofsV2) expected = proofsV2.NodeList() } - if err := expectResponse(server.peer.app, ProofsV2Msg, 42, testBufLimit, expected); err != nil { + if err := expectResponse(server.peer.app, protocol.ProofsV2Msg, 42, testBufLimit, expected); err != nil { t.Errorf("codes mismatch: %v", err) } } @@ -415,7 +416,7 @@ func testGetStaleProof(t *testing.T, protocol int) { func TestGetCHTProofsLes2(t *testing.T) { testGetCHTProofs(t, 2) } func TestGetCHTProofsLes3(t *testing.T) { testGetCHTProofs(t, 3) } -func testGetCHTProofs(t *testing.T, protocol int) { +func testGetCHTProofs(t *testing.T, p int) { config := light.TestServerIndexerConfig waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) { @@ -427,7 +428,7 @@ func testGetCHTProofs(t *testing.T, protocol int) { time.Sleep(10 * time.Millisecond) } } - server, tearDown := newServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, false, true, 0) + server, tearDown := newServerEnv(t, int(config.ChtSize+config.ChtConfirms), p, waitIndexers, false, true, 0) defer tearDown() bc := server.handler.blockchain @@ -439,23 +440,23 @@ func testGetCHTProofs(t *testing.T, protocol int) { key := make([]byte, 8) binary.BigEndian.PutUint64(key, config.ChtSize-1) - proofsV2 := HelperTrieResps{ + proofsV2 := protocol.HelperTrieResponse{ AuxData: [][]byte{rlp}, } root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(config.ChtSize-1).Hash()) trie, _ := trie.New(root, trie.NewDatabase(rawdb.NewTable(server.db, light.ChtTablePrefix))) trie.Prove(key, 0, &proofsV2.Proofs) // Assemble the requests for the different protocols - requestsV2 := []HelperTrieReq{{ - Type: htCanonical, - TrieIdx: 0, - Key: key, - AuxReq: auxHeader, + requestsV2 := []protocol.HelperTrieRequest{{ + Type: protocol.HelperTrieCHT, + TrieIndex: 0, + Key: key, + AuxType: protocol.AuxHeader, }} // Send the proof request and verify the response - cost := server.peer.peer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2)) - sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2) - if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, len(requestsV2)) + sendRequest(server.peer.app, protocol.GetHelperTrieProofsMsg, 42, cost, requestsV2) + if err := expectResponse(server.peer.app, protocol.HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil { t.Errorf("proofs mismatch: %v", err) } } @@ -464,7 +465,7 @@ func TestGetBloombitsProofsLes2(t *testing.T) { testGetBloombitsProofs(t, 2) } func TestGetBloombitsProofsLes3(t *testing.T) { testGetBloombitsProofs(t, 3) } // Tests that bloombits proofs can be correctly retrieved. -func testGetBloombitsProofs(t *testing.T, protocol int) { +func testGetBloombitsProofs(t *testing.T, p int) { config := light.TestServerIndexerConfig waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) { @@ -476,7 +477,7 @@ func testGetBloombitsProofs(t *testing.T, protocol int) { time.Sleep(10 * time.Millisecond) } } - server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirms), protocol, waitIndexers, false, true, 0) + server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirms), p, waitIndexers, false, true, 0) defer tearDown() bc := server.handler.blockchain @@ -490,21 +491,21 @@ func testGetBloombitsProofs(t *testing.T, protocol int) { // Only the first bloom section has data. binary.BigEndian.PutUint64(key[2:], 0) - requests := []HelperTrieReq{{ - Type: htBloomBits, - TrieIdx: 0, - Key: key, + requests := []protocol.HelperTrieRequest{{ + Type: protocol.HelperTrieBloomTrie, + TrieIndex: 0, + Key: key, }} - var proofs HelperTrieResps + var proofs protocol.HelperTrieResponse root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomTrieSize-1).Hash()) trie, _ := trie.New(root, trie.NewDatabase(rawdb.NewTable(server.db, light.BloomTrieTablePrefix))) trie.Prove(key, 0, &proofs.Proofs) // Send the proof request and verify the response - cost := server.peer.peer.GetRequestCost(GetHelperTrieProofsMsg, len(requests)) - sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, cost, requests) - if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil { + cost := server.peer.peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, len(requests)) + sendRequest(server.peer.app, protocol.GetHelperTrieProofsMsg, 42, cost, requests) + if err := expectResponse(server.peer.app, protocol.HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil { t.Errorf("bit %d: proofs mismatch: %v", bit, err) } } @@ -513,8 +514,8 @@ func testGetBloombitsProofs(t *testing.T, protocol int) { func TestTransactionStatusLes2(t *testing.T) { testTransactionStatus(t, 2) } func TestTransactionStatusLes3(t *testing.T) { testTransactionStatus(t, 3) } -func testTransactionStatus(t *testing.T, protocol int) { - server, tearDown := newServerEnv(t, 0, protocol, nil, false, true, 0) +func testTransactionStatus(t *testing.T, p int) { + server, tearDown := newServerEnv(t, 0, p, nil, false, true, 0) defer tearDown() server.handler.addTxsSync = true @@ -525,13 +526,13 @@ func testTransactionStatus(t *testing.T, protocol int) { test := func(tx *types.Transaction, send bool, expStatus light.TxStatus) { reqID++ if send { - cost := server.peer.peer.GetRequestCost(SendTxV2Msg, 1) - sendRequest(server.peer.app, SendTxV2Msg, reqID, cost, types.Transactions{tx}) + cost := server.peer.peer.GetRequestCost(protocol.SendTxV2Msg, 1) + sendRequest(server.peer.app, protocol.SendTxV2Msg, reqID, cost, types.Transactions{tx}) } else { - cost := server.peer.peer.GetRequestCost(GetTxStatusMsg, 1) - sendRequest(server.peer.app, GetTxStatusMsg, reqID, cost, []common.Hash{tx.Hash()}) + cost := server.peer.peer.GetRequestCost(protocol.GetTxStatusMsg, 1) + sendRequest(server.peer.app, protocol.GetTxStatusMsg, reqID, cost, []common.Hash{tx.Hash()}) } - if err := expectResponse(server.peer.app, TxStatusMsg, reqID, testBufLimit, []light.TxStatus{expStatus}); err != nil { + if err := expectResponse(server.peer.app, protocol.TxStatusMsg, reqID, testBufLimit, []light.TxStatus{expStatus}); err != nil { t.Errorf("transaction status mismatch") } } @@ -620,14 +621,14 @@ func TestStopResumeLes3(t *testing.T) { header := server.handler.blockchain.CurrentHeader() req := func() { reqID++ - sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, testCost, &getBlockHeadersData{Origin: hashOrNumber{Hash: header.Hash()}, Amount: 1}) + sendRequest(server.peer.app, protocol.GetBlockHeadersMsg, reqID, testCost, &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: header.Hash()}, Amount: 1}) } for i := 1; i <= 5; i++ { // send requests while we still have enough buffer and expect a response for expBuf >= testCost { req() expBuf -= testCost - if err := expectResponse(server.peer.app, BlockHeadersMsg, reqID, expBuf, []*types.Header{header}); err != nil { + if err := expectResponse(server.peer.app, protocol.BlockHeadersMsg, reqID, expBuf, []*types.Header{header}); err != nil { t.Errorf("expected response and failed: %v", err) } } @@ -637,7 +638,7 @@ func TestStopResumeLes3(t *testing.T) { req() c-- } - if err := p2p.ExpectMsg(server.peer.app, StopMsg, nil); err != nil { + if err := p2p.ExpectMsg(server.peer.app, protocol.StopMsg, nil); err != nil { t.Errorf("expected StopMsg and failed: %v", err) } // wait until the buffer is recharged by half of the limit @@ -646,7 +647,7 @@ func TestStopResumeLes3(t *testing.T) { // expect a ResumeMsg with the partially recharged buffer value expBuf += testBufRecharge * wait - if err := p2p.ExpectMsg(server.peer.app, ResumeMsg, expBuf); err != nil { + if err := p2p.ExpectMsg(server.peer.app, protocol.ResumeMsg, expBuf); err != nil { t.Errorf("expected ResumeMsg and failed: %v", err) } } diff --git a/les/odr_requests.go b/les/odr_requests.go index 3c4dd7090f..66e103f753 100644 --- a/les/odr_requests.go +++ b/les/odr_requests.go @@ -26,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rlp" @@ -79,7 +80,7 @@ type BlockRequest light.BlockRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *BlockRequest) GetCost(peer *peer) uint64 { - return peer.GetRequestCost(GetBlockBodiesMsg, 1) + return peer.GetRequestCost(protocol.GetBlockBodiesMsg, 1) } // CanSend tells if a certain peer is suitable for serving the given request @@ -135,7 +136,7 @@ type ReceiptsRequest light.ReceiptsRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *ReceiptsRequest) GetCost(peer *peer) uint64 { - return peer.GetRequestCost(GetReceiptsMsg, 1) + return peer.GetRequestCost(protocol.GetReceiptsMsg, 1) } // CanSend tells if a certain peer is suitable for serving the given request @@ -180,19 +181,13 @@ func (r *ReceiptsRequest) Validate(db ethdb.Database, msg *Msg) error { return nil } -type ProofReq struct { - BHash common.Hash - AccKey, Key []byte - FromLevel uint -} - // ODR request type for state/storage trie entries, see LesOdrRequest interface type TrieRequest light.TrieRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *TrieRequest) GetCost(peer *peer) uint64 { - return peer.GetRequestCost(GetProofsV2Msg, 1) + return peer.GetRequestCost(protocol.GetProofsV2Msg, 1) } // CanSend tells if a certain peer is suitable for serving the given request @@ -203,12 +198,12 @@ func (r *TrieRequest) CanSend(peer *peer) bool { // Request sends an ODR request to the LES network (implementation of LesOdrRequest) func (r *TrieRequest) Request(reqID uint64, peer *peer) error { peer.Log().Debug("Requesting trie proof", "root", r.Id.Root, "key", r.Key) - req := ProofReq{ - BHash: r.Id.BlockHash, - AccKey: r.Id.AccKey, - Key: r.Key, + req := protocol.TrieProofRequest{ + BlockHash: r.Id.BlockHash, + Account: r.Id.AccKey, + Key: r.Key, } - return peer.RequestProofs(reqID, r.GetCost(peer), []ProofReq{req}) + return peer.RequestProofs(reqID, r.GetCost(peer), []protocol.TrieProofRequest{req}) } // Valid processes an ODR request reply message from the LES network @@ -235,18 +230,13 @@ func (r *TrieRequest) Validate(db ethdb.Database, msg *Msg) error { return nil } -type CodeReq struct { - BHash common.Hash - AccKey []byte -} - // ODR request type for node data (used for retrieving contract code), see LesOdrRequest interface type CodeRequest light.CodeRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *CodeRequest) GetCost(peer *peer) uint64 { - return peer.GetRequestCost(GetCodeMsg, 1) + return peer.GetRequestCost(protocol.GetCodeMsg, 1) } // CanSend tells if a certain peer is suitable for serving the given request @@ -257,11 +247,11 @@ func (r *CodeRequest) CanSend(peer *peer) bool { // Request sends an ODR request to the LES network (implementation of LesOdrRequest) func (r *CodeRequest) Request(reqID uint64, peer *peer) error { peer.Log().Debug("Requesting code data", "hash", r.Hash) - req := CodeReq{ - BHash: r.Id.BlockHash, - AccKey: r.Id.AccKey, + req := protocol.CodeRequest{ + BlockHash: r.Id.BlockHash, + Account: r.Id.AccKey, } - return peer.RequestCode(reqID, r.GetCost(peer), []CodeReq{req}) + return peer.RequestCode(reqID, r.GetCost(peer), []protocol.CodeRequest{req}) } // Valid processes an ODR request reply message from the LES network @@ -288,36 +278,13 @@ func (r *CodeRequest) Validate(db ethdb.Database, msg *Msg) error { return nil } -const ( - // helper trie type constants - htCanonical = iota // Canonical hash trie - htBloomBits // BloomBits trie - - // applicable for all helper trie requests - auxRoot = 1 - // applicable for htCanonical - auxHeader = 2 -) - -type HelperTrieReq struct { - Type uint - TrieIdx uint64 - Key []byte - FromLevel, AuxReq uint -} - -type HelperTrieResps struct { // describes all responses, not just a single one - Proofs light.NodeList - AuxData [][]byte -} - // ODR request type for requesting headers by Canonical Hash Trie, see LesOdrRequest interface type ChtRequest light.ChtRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *ChtRequest) GetCost(peer *peer) uint64 { - return peer.GetRequestCost(GetHelperTrieProofsMsg, 1) + return peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, 1) } // CanSend tells if a certain peer is suitable for serving the given request @@ -337,13 +304,13 @@ func (r *ChtRequest) Request(reqID uint64, peer *peer) error { peer.Log().Debug("Requesting CHT", "cht", r.ChtNum, "block", r.BlockNum) var encNum [8]byte binary.BigEndian.PutUint64(encNum[:], r.BlockNum) - req := HelperTrieReq{ - Type: htCanonical, - TrieIdx: r.ChtNum, - Key: encNum[:], - AuxReq: auxHeader, + req := protocol.HelperTrieRequest{ + Type: protocol.HelperTrieCHT, + TrieIndex: r.ChtNum, + Key: encNum[:], + AuxType: protocol.AuxHeader, } - return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req}) + return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []protocol.HelperTrieRequest{req}) } // Valid processes an ODR request reply message from the LES network @@ -355,7 +322,7 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error { if msg.MsgType != MsgHelperTrieProofs { return errInvalidMessageType } - resp := msg.Obj.(HelperTrieResps) + resp := msg.Obj.(protocol.HelperTrieResponse) if len(resp.AuxData) != 1 { return errInvalidEntryCount } @@ -404,17 +371,13 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error { return nil } -type BloomReq struct { - BloomTrieNum, BitIdx, SectionIndex, FromLevel uint64 -} - // ODR request type for requesting headers by Canonical Hash Trie, see LesOdrRequest interface type BloomRequest light.BloomRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *BloomRequest) GetCost(peer *peer) uint64 { - return peer.GetRequestCost(GetHelperTrieProofsMsg, len(r.SectionIndexList)) + return peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, len(r.SectionIndexList)) } // CanSend tells if a certain peer is suitable for serving the given request @@ -422,7 +385,7 @@ func (r *BloomRequest) CanSend(peer *peer) bool { peer.lock.RLock() defer peer.lock.RUnlock() - if peer.version < lpv2 { + if peer.version < protocol.Lpv2 { return false } return peer.headInfo.Number >= r.Config.BloomTrieConfirms && r.BloomTrieNum <= (peer.headInfo.Number-r.Config.BloomTrieConfirms)/r.Config.BloomTrieSize @@ -431,17 +394,17 @@ func (r *BloomRequest) CanSend(peer *peer) bool { // Request sends an ODR request to the LES network (implementation of LesOdrRequest) func (r *BloomRequest) Request(reqID uint64, peer *peer) error { peer.Log().Debug("Requesting BloomBits", "bloomTrie", r.BloomTrieNum, "bitIdx", r.BitIdx, "sections", r.SectionIndexList) - reqs := make([]HelperTrieReq, len(r.SectionIndexList)) + reqs := make([]protocol.HelperTrieRequest, len(r.SectionIndexList)) var encNumber [10]byte binary.BigEndian.PutUint16(encNumber[:2], uint16(r.BitIdx)) for i, sectionIdx := range r.SectionIndexList { binary.BigEndian.PutUint64(encNumber[2:], sectionIdx) - reqs[i] = HelperTrieReq{ - Type: htBloomBits, - TrieIdx: r.BloomTrieNum, - Key: common.CopyBytes(encNumber[:]), + reqs[i] = protocol.HelperTrieRequest{ + Type: protocol.HelperTrieBloomTrie, + TrieIndex: r.BloomTrieNum, + Key: common.CopyBytes(encNumber[:]), } } return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), reqs) @@ -457,7 +420,7 @@ func (r *BloomRequest) Validate(db ethdb.Database, msg *Msg) error { if msg.MsgType != MsgHelperTrieProofs { return errInvalidMessageType } - resps := msg.Obj.(HelperTrieResps) + resps := msg.Obj.(protocol.HelperTrieResponse) proofs := resps.Proofs nodeSet := proofs.NodeSet() reads := &readTraceDB{db: nodeSet} @@ -490,12 +453,12 @@ type TxStatusRequest light.TxStatusRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *TxStatusRequest) GetCost(peer *peer) uint64 { - return peer.GetRequestCost(GetTxStatusMsg, len(r.Hashes)) + return peer.GetRequestCost(protocol.GetTxStatusMsg, len(r.Hashes)) } // CanSend tells if a certain peer is suitable for serving the given request func (r *TxStatusRequest) CanSend(peer *peer) bool { - return peer.version >= lpv2 + return peer.version >= protocol.Lpv2 } // Request sends an ODR request to the LES network (implementation of LesOdrRequest) diff --git a/les/peer.go b/les/peer.go index e1bf273524..817fb8f564 100644 --- a/les/peer.go +++ b/les/peer.go @@ -32,6 +32,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/les/utilities" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/p2p" @@ -89,7 +90,7 @@ type peer struct { id string - headInfo *announceData + headInfo *protocol.Announcement lock sync.RWMutex sendQueue *utilities.ExecQueue @@ -112,7 +113,7 @@ type peer struct { fcClient *flowcontrol.ClientNode // nil if the peer is server only fcServer *flowcontrol.ServerNode // nil if the peer is client only fcParams flowcontrol.ServerParams - fcCosts requestCostTable + fcCosts protocol.RequestCostTable trusted, server bool onlyAnnounce bool @@ -179,7 +180,7 @@ func (p *peer) rejectUpdate(size uint64) bool { // region. The client is also notified about being frozen/unfrozen with a Stop/Resume // message. func (p *peer) freezeClient() { - if p.version < lpv3 { + if p.version < protocol.Lpv3 { // if Stop/Resume is not supported then just drop the peer after setting // its frozen status permanently atomic.StoreUint32(&p.frozen, 1) @@ -258,11 +259,11 @@ func (p *peer) HeadAndTd() (hash common.Hash, td *big.Int) { return hash, p.headInfo.Td } -func (p *peer) headBlockInfo() blockInfo { +func (p *peer) headBlockInfo() protocol.HeadHeader { p.lock.RLock() defer p.lock.RUnlock() - return blockInfo{Hash: p.headInfo.Hash, Number: p.headInfo.Number, Td: p.headInfo.Td} + return protocol.HeadHeader{Hash: p.headInfo.Hash, Number: p.headInfo.Number, Td: p.headInfo.Td} } // Td retrieves the current total difficulty of a peer. @@ -286,10 +287,10 @@ func (p *peer) updateCapacity(cap uint64) { p.fcParams = flowcontrol.ServerParams{MinRecharge: cap, BufLimit: cap * bufLimitRatio} p.fcClient.UpdateParams(p.fcParams) - var kvList keyValueList - kvList = kvList.add("flowControl/MRR", cap) - kvList = kvList.add("flowControl/BL", cap*bufLimitRatio) - p.queueSend(func() { p.SendAnnounce(announceData{Update: kvList}) }) + var kvList protocol.KeyValueList + kvList = kvList.Add("flowControl/MRR", cap) + kvList = kvList.Add("flowControl/BL", cap*bufLimitRatio) + p.queueSend(func() { p.SendAnnounce(protocol.Announcement{Update: kvList}) }) } func (p *peer) responseID() uint64 { @@ -336,7 +337,7 @@ func (p *peer) GetRequestCost(msgcode uint64, amount int) uint64 { if costs == nil { return 0 } - cost := costs.baseCost + costs.reqCost*uint64(amount) + cost := costs.BaseCost + costs.ReqCost*uint64(amount) if cost > p.fcParams.BufLimit { cost = p.fcParams.BufLimit } @@ -347,12 +348,12 @@ func (p *peer) GetTxRelayCost(amount, size int) uint64 { p.lock.RLock() defer p.lock.RUnlock() - costs := p.fcCosts[SendTxV2Msg] + costs := p.fcCosts[protocol.SendTxV2Msg] if costs == nil { return 0 } - cost := costs.baseCost + costs.reqCost*uint64(amount) - sizeCost := costs.baseCost + costs.reqCost*uint64(size)/txSizeCostLimit + cost := costs.BaseCost + costs.ReqCost*uint64(amount) + sizeCost := costs.BaseCost + costs.ReqCost*uint64(size)/txSizeCostLimit if sizeCost > cost { cost = sizeCost } @@ -385,185 +386,144 @@ func (p *peer) HasBlock(hash common.Hash, number uint64, hasState bool) bool { // SendAnnounce announces the availability of a number of blocks through // a hash notification. -func (p *peer) SendAnnounce(request announceData) error { - return p2p.Send(p.rw, AnnounceMsg, request) +func (p *peer) SendAnnounce(request protocol.Announcement) error { + return p2p.Send(p.rw, protocol.AnnounceMsg, request) } // SendStop notifies the client about being in frozen state func (p *peer) SendStop() error { - return p2p.Send(p.rw, StopMsg, struct{}{}) + return p2p.Send(p.rw, protocol.StopMsg, struct{}{}) } // SendResume notifies the client about getting out of frozen state func (p *peer) SendResume(bv uint64) error { - return p2p.Send(p.rw, ResumeMsg, bv) + return p2p.Send(p.rw, protocol.ResumeMsg, bv) } // ReplyBlockHeaders creates a reply with a batch of block headers func (p *peer) ReplyBlockHeaders(reqID uint64, headers []*types.Header) *reply { data, _ := rlp.EncodeToBytes(headers) - return &reply{p.rw, BlockHeadersMsg, reqID, data} + return &reply{p.rw, protocol.BlockHeadersMsg, reqID, data} } // ReplyBlockBodiesRLP creates a reply with a batch of block contents from // an already RLP encoded format. func (p *peer) ReplyBlockBodiesRLP(reqID uint64, bodies []rlp.RawValue) *reply { data, _ := rlp.EncodeToBytes(bodies) - return &reply{p.rw, BlockBodiesMsg, reqID, data} + return &reply{p.rw, protocol.BlockBodiesMsg, reqID, data} } // ReplyCode creates a reply with a batch of arbitrary internal data, corresponding to the // hashes requested. func (p *peer) ReplyCode(reqID uint64, codes [][]byte) *reply { data, _ := rlp.EncodeToBytes(codes) - return &reply{p.rw, CodeMsg, reqID, data} + return &reply{p.rw, protocol.CodeMsg, reqID, data} } // ReplyReceiptsRLP creates a reply with a batch of transaction receipts, corresponding to the // ones requested from an already RLP encoded format. func (p *peer) ReplyReceiptsRLP(reqID uint64, receipts []rlp.RawValue) *reply { data, _ := rlp.EncodeToBytes(receipts) - return &reply{p.rw, ReceiptsMsg, reqID, data} + return &reply{p.rw, protocol.ReceiptsMsg, reqID, data} } // ReplyProofsV2 creates a reply with a batch of merkle proofs, corresponding to the ones requested. func (p *peer) ReplyProofsV2(reqID uint64, proofs light.NodeList) *reply { data, _ := rlp.EncodeToBytes(proofs) - return &reply{p.rw, ProofsV2Msg, reqID, data} + return &reply{p.rw, protocol.ProofsV2Msg, reqID, data} } // ReplyHelperTrieProofs creates a reply with a batch of HelperTrie proofs, corresponding to the ones requested. -func (p *peer) ReplyHelperTrieProofs(reqID uint64, resp HelperTrieResps) *reply { +func (p *peer) ReplyHelperTrieProofs(reqID uint64, resp protocol.HelperTrieResponse) *reply { data, _ := rlp.EncodeToBytes(resp) - return &reply{p.rw, HelperTrieProofsMsg, reqID, data} + return &reply{p.rw, protocol.HelperTrieProofsMsg, reqID, data} } // ReplyTxStatus creates a reply with a batch of transaction status records, corresponding to the ones requested. func (p *peer) ReplyTxStatus(reqID uint64, stats []light.TxStatus) *reply { data, _ := rlp.EncodeToBytes(stats) - return &reply{p.rw, TxStatusMsg, reqID, data} + return &reply{p.rw, protocol.TxStatusMsg, reqID, data} } // RequestHeadersByHash fetches a batch of blocks' headers corresponding to the // specified header query, based on the hash of an origin block. func (p *peer) RequestHeadersByHash(reqID, cost uint64, origin common.Hash, amount int, skip int, reverse bool) error { p.Log().Debug("Fetching batch of headers", "count", amount, "fromhash", origin, "skip", skip, "reverse", reverse) - return sendRequest(p.rw, GetBlockHeadersMsg, reqID, cost, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) + return sendRequest(p.rw, protocol.GetBlockHeadersMsg, reqID, cost, &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) } // RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the // specified header query, based on the number of an origin block. func (p *peer) RequestHeadersByNumber(reqID, cost, origin uint64, amount int, skip int, reverse bool) error { p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse) - return sendRequest(p.rw, GetBlockHeadersMsg, reqID, cost, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) + return sendRequest(p.rw, protocol.GetBlockHeadersMsg, reqID, cost, &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) } // RequestBodies fetches a batch of blocks' bodies corresponding to the hashes // specified. func (p *peer) RequestBodies(reqID, cost uint64, hashes []common.Hash) error { p.Log().Debug("Fetching batch of block bodies", "count", len(hashes)) - return sendRequest(p.rw, GetBlockBodiesMsg, reqID, cost, hashes) + return sendRequest(p.rw, protocol.GetBlockBodiesMsg, reqID, cost, hashes) } // RequestCode fetches a batch of arbitrary data from a node's known state // data, corresponding to the specified hashes. -func (p *peer) RequestCode(reqID, cost uint64, reqs []CodeReq) error { +func (p *peer) RequestCode(reqID, cost uint64, reqs []protocol.CodeRequest) error { p.Log().Debug("Fetching batch of codes", "count", len(reqs)) - return sendRequest(p.rw, GetCodeMsg, reqID, cost, reqs) + return sendRequest(p.rw, protocol.GetCodeMsg, reqID, cost, reqs) } // RequestReceipts fetches a batch of transaction receipts from a remote node. func (p *peer) RequestReceipts(reqID, cost uint64, hashes []common.Hash) error { p.Log().Debug("Fetching batch of receipts", "count", len(hashes)) - return sendRequest(p.rw, GetReceiptsMsg, reqID, cost, hashes) + return sendRequest(p.rw, protocol.GetReceiptsMsg, reqID, cost, hashes) } // RequestProofs fetches a batch of merkle proofs from a remote node. -func (p *peer) RequestProofs(reqID, cost uint64, reqs []ProofReq) error { +func (p *peer) RequestProofs(reqID, cost uint64, reqs []protocol.TrieProofRequest) error { p.Log().Debug("Fetching batch of proofs", "count", len(reqs)) - return sendRequest(p.rw, GetProofsV2Msg, reqID, cost, reqs) + return sendRequest(p.rw, protocol.GetProofsV2Msg, reqID, cost, reqs) } // RequestHelperTrieProofs fetches a batch of HelperTrie merkle proofs from a remote node. -func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []HelperTrieReq) error { +func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []protocol.HelperTrieRequest) error { p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs)) - return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, cost, reqs) + return sendRequest(p.rw, protocol.GetHelperTrieProofsMsg, reqID, cost, reqs) } // RequestTxStatus fetches a batch of transaction status records from a remote node. func (p *peer) RequestTxStatus(reqID, cost uint64, txHashes []common.Hash) error { p.Log().Debug("Requesting transaction status", "count", len(txHashes)) - return sendRequest(p.rw, GetTxStatusMsg, reqID, cost, txHashes) + return sendRequest(p.rw, protocol.GetTxStatusMsg, reqID, cost, txHashes) } // SendTxStatus creates a reply with a batch of transactions to be added to the remote transaction pool. func (p *peer) SendTxs(reqID, cost uint64, txs rlp.RawValue) error { p.Log().Debug("Sending batch of transactions", "size", len(txs)) - return sendRequest(p.rw, SendTxV2Msg, reqID, cost, txs) + return sendRequest(p.rw, protocol.SendTxV2Msg, reqID, cost, txs) } -type keyValueEntry struct { - Key string - Value rlp.RawValue -} -type keyValueList []keyValueEntry -type keyValueMap map[string]rlp.RawValue - -func (l keyValueList) add(key string, val interface{}) keyValueList { - var entry keyValueEntry - entry.Key = key - if val == nil { - val = uint64(0) - } - enc, err := rlp.EncodeToBytes(val) - if err == nil { - entry.Value = enc - } - return append(l, entry) -} - -func (l keyValueList) decode() (keyValueMap, uint64) { - m := make(keyValueMap) - var size uint64 - for _, entry := range l { - m[entry.Key] = entry.Value - size += uint64(len(entry.Key)) + uint64(len(entry.Value)) + 8 - } - return m, size -} - -func (m keyValueMap) get(key string, val interface{}) error { - enc, ok := m[key] - if !ok { - return errResp(ErrMissingKey, "%s", key) - } - if val == nil { - return nil - } - return rlp.DecodeBytes(enc, val) -} - -func (p *peer) sendReceiveHandshake(sendList keyValueList) (keyValueList, error) { +func (p *peer) sendReceiveHandshake(sendList protocol.KeyValueList) (protocol.KeyValueList, error) { // Send out own handshake in a new thread errc := make(chan error, 1) go func() { - errc <- p2p.Send(p.rw, StatusMsg, sendList) + errc <- p2p.Send(p.rw, protocol.StatusMsg, sendList) }() // In the mean time retrieve the remote status message msg, err := p.rw.ReadMsg() if err != nil { return nil, err } - if msg.Code != StatusMsg { - return nil, errResp(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg) + if msg.Code != protocol.StatusMsg { + return nil, protocol.ErrResp(protocol.ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, protocol.StatusMsg) } - if msg.Size > ProtocolMaxMsgSize { - return nil, errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + if msg.Size > protocol.ProtocolMaxMsgSize { + return nil, protocol.ErrResp(protocol.ErrMsgTooLarge, "%v > %v", msg.Size, protocol.ProtocolMaxMsgSize) } // Decode the handshake - var recvList keyValueList + var recvList protocol.KeyValueList if err := msg.Decode(&recvList); err != nil { - return nil, errResp(ErrDecode, "msg %v: %v", msg, err) + return nil, protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } if err := <-errc; err != nil { return nil, err @@ -577,21 +537,21 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis p.lock.Lock() defer p.lock.Unlock() - var send keyValueList + var send protocol.KeyValueList // Add some basic handshake fields - send = send.add("protocolVersion", uint64(p.version)) - send = send.add("networkId", p.network) - send = send.add("headTd", td) - send = send.add("headHash", head) - send = send.add("headNum", headNum) - send = send.add("genesisHash", genesis) + send = send.Add("protocolVersion", uint64(p.version)) + send = send.Add("networkId", p.network) + send = send.Add("headTd", td) + send = send.Add("headHash", head) + send = send.Add("headNum", headNum) + send = send.Add("genesisHash", genesis) if server != nil { // Add some information which services server can offer. if !server.config.UltraLightOnlyAnnounce { - send = send.add("serveHeaders", nil) - send = send.add("serveChainSince", uint64(0)) - send = send.add("serveStateSince", uint64(0)) + send = send.Add("serveHeaders", nil) + send = send.Add("serveChainSince", uint64(0)) + send = send.Add("serveStateSince", uint64(0)) // If local ethereum node is running in archive mode, advertise ourselves we have // all version state data. Otherwise only recent state is available. @@ -599,20 +559,20 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis if server.archiveMode { stateRecent = 0 } - send = send.add("serveRecentState", stateRecent) - send = send.add("txRelay", nil) + send = send.Add("serveRecentState", stateRecent) + send = send.Add("txRelay", nil) } - send = send.add("flowControl/BL", server.defParams.BufLimit) - send = send.add("flowControl/MRR", server.defParams.MinRecharge) + send = send.Add("flowControl/BL", server.defParams.BufLimit) + send = send.Add("flowControl/MRR", server.defParams.MinRecharge) - var costList RequestCostList + var costList protocol.RequestCostList if server.costTracker.testCostList != nil { costList = server.costTracker.testCostList } else { costList = server.costTracker.makeCostList(server.costTracker.globalFactor()) } - send = send.add("flowControl/MRC", costList) - p.fcCosts = costList.decode(ProtocolLengths[uint(p.version)]) + send = send.Add("flowControl/MRC", costList) + p.fcCosts = costList.ToTable(protocol.ProtocolLengths[uint(p.version)]) p.fcParams = server.defParams // Add advertised checkpoint and register block height which @@ -620,8 +580,8 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis if server.oracle != nil && server.oracle.IsRunning() { cp, height := server.oracle.StableCheckpoint() if cp != nil { - send = send.add("checkpoint/value", cp) - send = send.add("checkpoint/registerHeight", height) + send = send.Add("checkpoint/value", cp) + send = send.Add("checkpoint/registerHeight", height) } } } else { @@ -630,130 +590,136 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis if p.trusted { p.announceType = announceTypeSigned } - send = send.add("announceType", p.announceType) + send = send.Add("announceType", p.announceType) } recvList, err := p.sendReceiveHandshake(send) if err != nil { return err } - recv, size := recvList.decode() + recv, size := recvList.ToMap() if p.rejectUpdate(size) { - return errResp(ErrRequestRejected, "") + return protocol.ErrResp(protocol.ErrRequestRejected, "") } var rGenesis, rHash common.Hash var rVersion, rNetwork, rNum uint64 var rTd *big.Int - if err := recv.get("protocolVersion", &rVersion); err != nil { + if err := recv.Get("protocolVersion", &rVersion); err != nil { return err } - if err := recv.get("networkId", &rNetwork); err != nil { + if err := recv.Get("networkId", &rNetwork); err != nil { return err } - if err := recv.get("headTd", &rTd); err != nil { + if err := recv.Get("headTd", &rTd); err != nil { return err } - if err := recv.get("headHash", &rHash); err != nil { + if err := recv.Get("headHash", &rHash); err != nil { return err } - if err := recv.get("headNum", &rNum); err != nil { + if err := recv.Get("headNum", &rNum); err != nil { return err } - if err := recv.get("genesisHash", &rGenesis); err != nil { + if err := recv.Get("genesisHash", &rGenesis); err != nil { return err } if rGenesis != genesis { - return errResp(ErrGenesisBlockMismatch, "%x (!= %x)", rGenesis[:8], genesis[:8]) + return protocol.ErrResp(protocol.ErrGenesisBlockMismatch, "%x (!= %x)", rGenesis[:8], genesis[:8]) } if rNetwork != p.network { - return errResp(ErrNetworkIdMismatch, "%d (!= %d)", rNetwork, p.network) + return protocol.ErrResp(protocol.ErrNetworkIdMismatch, "%d (!= %d)", rNetwork, p.network) } if int(rVersion) != p.version { - return errResp(ErrProtocolVersionMismatch, "%d (!= %d)", rVersion, p.version) + return protocol.ErrResp(protocol.ErrProtocolVersionMismatch, "%d (!= %d)", rVersion, p.version) } if server != nil { - p.server = recv.get("flowControl/MRR", nil) == nil + p.server = recv.Get("flowControl/MRR", nil) == nil if p.server { p.announceType = announceTypeNone // connected to another server, send no messages } else { - if recv.get("announceType", &p.announceType) != nil { + if recv.Get("announceType", &p.announceType) != nil { // set default announceType on server side p.announceType = announceTypeSimple } p.fcClient = flowcontrol.NewClientNode(server.fcManager, server.defParams) } } else { - if recv.get("serveChainSince", &p.chainSince) != nil { + if recv.Get("serveChainSince", &p.chainSince) != nil { p.onlyAnnounce = true } - if recv.get("serveRecentChain", &p.chainRecent) != nil { + if recv.Get("serveRecentChain", &p.chainRecent) != nil { p.chainRecent = 0 } - if recv.get("serveStateSince", &p.stateSince) != nil { + if recv.Get("serveStateSince", &p.stateSince) != nil { p.onlyAnnounce = true } - if recv.get("serveRecentState", &p.stateRecent) != nil { + if recv.Get("serveRecentState", &p.stateRecent) != nil { p.stateRecent = 0 } - if recv.get("txRelay", nil) != nil { + if recv.Get("txRelay", nil) != nil { p.onlyAnnounce = true } if p.onlyAnnounce && !p.trusted { - return errResp(ErrUselessPeer, "peer cannot serve requests") + return protocol.ErrResp(protocol.ErrUselessPeer, "peer cannot serve requests") } var sParams flowcontrol.ServerParams - if err := recv.get("flowControl/BL", &sParams.BufLimit); err != nil { + if err := recv.Get("flowControl/BL", &sParams.BufLimit); err != nil { return err } - if err := recv.get("flowControl/MRR", &sParams.MinRecharge); err != nil { + if err := recv.Get("flowControl/MRR", &sParams.MinRecharge); err != nil { return err } - var MRC RequestCostList - if err := recv.get("flowControl/MRC", &MRC); err != nil { + var MRC protocol.RequestCostList + if err := recv.Get("flowControl/MRC", &MRC); err != nil { return err } p.fcParams = sParams p.fcServer = flowcontrol.NewServerNode(sParams, &mclock.System{}) - p.fcCosts = MRC.decode(ProtocolLengths[uint(p.version)]) + p.fcCosts = MRC.ToTable(protocol.ProtocolLengths[uint(p.version)]) - recv.get("checkpoint/value", &p.checkpoint) - recv.get("checkpoint/registerHeight", &p.checkpointNumber) + recv.Get("checkpoint/value", &p.checkpoint) + recv.Get("checkpoint/registerHeight", &p.checkpointNumber) if !p.onlyAnnounce { for msgCode := range reqAvgTimeCost { if p.fcCosts[msgCode] == nil { - return errResp(ErrUselessPeer, "peer does not support message %d", msgCode) + return protocol.ErrResp(protocol.ErrUselessPeer, "peer does not support message %d", msgCode) } } } p.server = true } - p.headInfo = &announceData{Td: rTd, Hash: rHash, Number: rNum} + p.headInfo = &protocol.Announcement{ + HeadHeader: protocol.HeadHeader{ + Td: rTd, + Hash: rHash, + Number: rNum, + }, + } return nil } // updateFlowControl updates the flow control parameters belonging to the server // node if the announced key/value set contains relevant fields -func (p *peer) updateFlowControl(update keyValueMap) { +func (p *peer) updateFlowControl(update protocol.KeyValueMap) { if p.fcServer == nil { return } // If any of the flow control params is nil, refuse to update. var params flowcontrol.ServerParams - if update.get("flowControl/BL", ¶ms.BufLimit) == nil && update.get("flowControl/MRR", ¶ms.MinRecharge) == nil { + if update.Get("flowControl/BL", ¶ms.BufLimit) == nil && update.Get("flowControl/MRR", ¶ms.MinRecharge) == nil { // todo can light client set a minimal acceptable flow control params? p.fcParams = params p.fcServer.UpdateParams(params) } - var MRC RequestCostList - if update.get("flowControl/MRC", &MRC) == nil { - costUpdate := MRC.decode(ProtocolLengths[uint(p.version)]) + var MRC protocol.RequestCostList + if update.Get("flowControl/MRC", &MRC) == nil { + costUpdate := MRC.ToTable(protocol.ProtocolLengths[uint(p.version)]) for code, cost := range costUpdate { p.fcCosts[code] = cost } diff --git a/les/peer_test.go b/les/peer_test.go index db74a052c1..2074bf5fc2 100644 --- a/les/peer_test.go +++ b/les/peer_test.go @@ -27,12 +27,13 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rlp" ) -const protocolVersion = lpv2 +const protocolVersion = protocol.Lpv2 var ( hash = common.HexToHash("deadbeef") @@ -59,10 +60,10 @@ func TestPeerHandshakeSetAnnounceTypeToAnnounceTypeSignedForTrustedPeer(t *testi version: protocolVersion, trusted: true, rw: &rwStub{ - WriteHook: func(recvList keyValueList) { - recv, _ := recvList.decode() + WriteHook: func(recvList protocol.KeyValueList) { + recv, _ := recvList.ToMap() var reqType uint64 - err := recv.get("announceType", &reqType) + err := recv.Get("announceType", &reqType) if err != nil { t.Fatal(err) } @@ -70,18 +71,18 @@ func TestPeerHandshakeSetAnnounceTypeToAnnounceTypeSignedForTrustedPeer(t *testi t.Fatal("Expected announceTypeSigned") } }, - ReadHook: func(l keyValueList) keyValueList { - l = l.add("serveHeaders", nil) - l = l.add("serveChainSince", uint64(0)) - l = l.add("serveStateSince", uint64(0)) - l = l.add("txRelay", nil) - l = l.add("flowControl/BL", uint64(0)) - l = l.add("flowControl/MRR", uint64(0)) - l = l.add("flowControl/MRC", testCostList(0)) + ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList { + l = l.Add("serveHeaders", nil) + l = l.Add("serveChainSince", uint64(0)) + l = l.Add("serveStateSince", uint64(0)) + l = l.Add("txRelay", nil) + l = l.Add("flowControl/BL", uint64(0)) + l = l.Add("flowControl/MRR", uint64(0)) + l = l.Add("flowControl/MRC", testCostList(0)) return l }, }, - network: NetworkId, + network: protocol.NetworkId, } err := p.Handshake(td, hash, headNum, genesis, nil) if err != nil { @@ -98,11 +99,11 @@ func TestPeerHandshakeAnnounceTypeSignedForTrustedPeersPeerNotInTrusted(t *testi Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}), version: protocolVersion, rw: &rwStub{ - WriteHook: func(recvList keyValueList) { + WriteHook: func(recvList protocol.KeyValueList) { // checking that ulc sends to peer allowedRequests=noRequests and announceType != announceTypeSigned - recv, _ := recvList.decode() + recv, _ := recvList.ToMap() var reqType uint64 - err := recv.get("announceType", &reqType) + err := recv.Get("announceType", &reqType) if err != nil { t.Fatal(err) } @@ -110,18 +111,18 @@ func TestPeerHandshakeAnnounceTypeSignedForTrustedPeersPeerNotInTrusted(t *testi t.Fatal("Expected not announceTypeSigned") } }, - ReadHook: func(l keyValueList) keyValueList { - l = l.add("serveHeaders", nil) - l = l.add("serveChainSince", uint64(0)) - l = l.add("serveStateSince", uint64(0)) - l = l.add("txRelay", nil) - l = l.add("flowControl/BL", uint64(0)) - l = l.add("flowControl/MRR", uint64(0)) - l = l.add("flowControl/MRC", testCostList(0)) + ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList { + l = l.Add("serveHeaders", nil) + l = l.Add("serveChainSince", uint64(0)) + l = l.Add("serveStateSince", uint64(0)) + l = l.Add("txRelay", nil) + l = l.Add("flowControl/BL", uint64(0)) + l = l.Add("flowControl/MRR", uint64(0)) + l = l.Add("flowControl/MRC", testCostList(0)) return l }, }, - network: NetworkId, + network: protocol.NetworkId, } err := p.Handshake(td, hash, headNum, genesis, nil) if err != nil { @@ -141,13 +142,13 @@ func TestPeerHandshakeDefaultAllRequests(t *testing.T) { Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}), version: protocolVersion, rw: &rwStub{ - ReadHook: func(l keyValueList) keyValueList { - l = l.add("announceType", uint64(announceTypeSigned)) - l = l.add("allowedRequests", uint64(0)) + ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList { + l = l.Add("announceType", uint64(announceTypeSigned)) + l = l.Add("allowedRequests", uint64(0)) return l }, }, - network: NetworkId, + network: protocol.NetworkId, } err := p.Handshake(td, hash, headNum, genesis, s) @@ -170,11 +171,11 @@ func TestPeerHandshakeServerSendOnlyAnnounceRequestsHeaders(t *testing.T) { Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}), version: protocolVersion, rw: &rwStub{ - ReadHook: func(l keyValueList) keyValueList { - l = l.add("announceType", uint64(announceTypeSigned)) + ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList { + l = l.Add("announceType", uint64(announceTypeSigned)) return l }, - WriteHook: func(l keyValueList) { + WriteHook: func(l protocol.KeyValueList) { for _, v := range l { if v.Key == "serveHeaders" || v.Key == "serveChainSince" || @@ -185,7 +186,7 @@ func TestPeerHandshakeServerSendOnlyAnnounceRequestsHeaders(t *testing.T) { } }, }, - network: NetworkId, + network: protocol.NetworkId, } err := p.Handshake(td, hash, headNum, genesis, s) @@ -200,17 +201,17 @@ func TestPeerHandshakeClientReceiveOnlyAnnounceRequestsHeaders(t *testing.T) { Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}), version: protocolVersion, rw: &rwStub{ - ReadHook: func(l keyValueList) keyValueList { - l = l.add("flowControl/BL", uint64(0)) - l = l.add("flowControl/MRR", uint64(0)) - l = l.add("flowControl/MRC", RequestCostList{}) + ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList { + l = l.Add("flowControl/BL", uint64(0)) + l = l.Add("flowControl/MRR", uint64(0)) + l = l.Add("flowControl/MRC", protocol.RequestCostList{}) - l = l.add("announceType", uint64(announceTypeSigned)) + l = l.Add("announceType", uint64(announceTypeSigned)) return l }, }, - network: NetworkId, + network: protocol.NetworkId, trusted: true, } @@ -231,15 +232,15 @@ func TestPeerHandshakeClientReturnErrorOnUselessPeer(t *testing.T) { Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}), version: protocolVersion, rw: &rwStub{ - ReadHook: func(l keyValueList) keyValueList { - l = l.add("flowControl/BL", uint64(0)) - l = l.add("flowControl/MRR", uint64(0)) - l = l.add("flowControl/MRC", RequestCostList{}) - l = l.add("announceType", uint64(announceTypeSigned)) + ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList { + l = l.Add("flowControl/BL", uint64(0)) + l = l.Add("flowControl/MRR", uint64(0)) + l = l.Add("flowControl/MRC", protocol.RequestCostList{}) + l = l.Add("announceType", uint64(announceTypeSigned)) return l }, }, - network: NetworkId, + network: protocol.NetworkId, } err := p.Handshake(td, hash, headNum, genesis, nil) @@ -264,18 +265,18 @@ func generateLesServer() *LesServer { } type rwStub struct { - ReadHook func(l keyValueList) keyValueList - WriteHook func(l keyValueList) + ReadHook func(l protocol.KeyValueList) protocol.KeyValueList + WriteHook func(l protocol.KeyValueList) } func (s *rwStub) ReadMsg() (p2p.Msg, error) { - payload := keyValueList{} - payload = payload.add("protocolVersion", uint64(protocolVersion)) - payload = payload.add("networkId", uint64(NetworkId)) - payload = payload.add("headTd", td) - payload = payload.add("headHash", hash) - payload = payload.add("headNum", headNum) - payload = payload.add("genesisHash", genesis) + payload := protocol.KeyValueList{} + payload = payload.Add("protocolVersion", uint64(protocolVersion)) + payload = payload.Add("networkId", uint64(protocol.NetworkId)) + payload = payload.Add("headTd", td) + payload = payload.Add("headHash", hash) + payload = payload.Add("headNum", headNum) + payload = payload.Add("genesisHash", genesis) if s.ReadHook != nil { payload = s.ReadHook(payload) @@ -291,7 +292,7 @@ func (s *rwStub) ReadMsg() (p2p.Msg, error) { } func (s *rwStub) WriteMsg(m p2p.Msg) error { - recvList := keyValueList{} + recvList := protocol.KeyValueList{} if err := m.Decode(&recvList); err != nil { return err } diff --git a/les/protocol.go b/les/protocol.go deleted file mode 100644 index 36af88aea6..0000000000 --- a/les/protocol.go +++ /dev/null @@ -1,237 +0,0 @@ -// Copyright 2016 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package les - -import ( - "crypto/ecdsa" - "errors" - "fmt" - "io" - "math/big" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/p2p/enode" - "github.com/ethereum/go-ethereum/rlp" -) - -// Constants to match up protocol versions and messages -const ( - lpv2 = 2 - lpv3 = 3 -) - -// Supported versions of the les protocol (first is primary) -var ( - ClientProtocolVersions = []uint{lpv2, lpv3} - ServerProtocolVersions = []uint{lpv2, lpv3} - AdvertiseProtocolVersions = []uint{lpv2} // clients are searching for the first advertised protocol in the list -) - -// Number of implemented message corresponding to different protocol versions. -var ProtocolLengths = map[uint]uint64{lpv2: 22, lpv3: 24} - -const ( - NetworkId = 1 - ProtocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message -) - -// les protocol message codes -const ( - // Protocol messages inherited from LPV1 - StatusMsg = 0x00 - AnnounceMsg = 0x01 - GetBlockHeadersMsg = 0x02 - BlockHeadersMsg = 0x03 - GetBlockBodiesMsg = 0x04 - BlockBodiesMsg = 0x05 - GetReceiptsMsg = 0x06 - ReceiptsMsg = 0x07 - GetCodeMsg = 0x0a - CodeMsg = 0x0b - // Protocol messages introduced in LPV2 - GetProofsV2Msg = 0x0f - ProofsV2Msg = 0x10 - GetHelperTrieProofsMsg = 0x11 - HelperTrieProofsMsg = 0x12 - SendTxV2Msg = 0x13 - GetTxStatusMsg = 0x14 - TxStatusMsg = 0x15 - // Protocol messages introduced in LPV3 - StopMsg = 0x16 - ResumeMsg = 0x17 -) - -type requestInfo struct { - name string - maxCount uint64 -} - -var requests = map[uint64]requestInfo{ - GetBlockHeadersMsg: {"GetBlockHeaders", MaxHeaderFetch}, - GetBlockBodiesMsg: {"GetBlockBodies", MaxBodyFetch}, - GetReceiptsMsg: {"GetReceipts", MaxReceiptFetch}, - GetCodeMsg: {"GetCode", MaxCodeFetch}, - GetProofsV2Msg: {"GetProofsV2", MaxProofsFetch}, - GetHelperTrieProofsMsg: {"GetHelperTrieProofs", MaxHelperTrieProofsFetch}, - SendTxV2Msg: {"SendTxV2", MaxTxSend}, - GetTxStatusMsg: {"GetTxStatus", MaxTxStatus}, -} - -type errCode int - -const ( - ErrMsgTooLarge = iota - ErrDecode - ErrInvalidMsgCode - ErrProtocolVersionMismatch - ErrNetworkIdMismatch - ErrGenesisBlockMismatch - ErrNoStatusMsg - ErrExtraStatusMsg - ErrSuspendedPeer - ErrUselessPeer - ErrRequestRejected - ErrUnexpectedResponse - ErrInvalidResponse - ErrTooManyTimeouts - ErrMissingKey -) - -func (e errCode) String() string { - return errorToString[int(e)] -} - -// XXX change once legacy code is out -var errorToString = map[int]string{ - ErrMsgTooLarge: "Message too long", - ErrDecode: "Invalid message", - ErrInvalidMsgCode: "Invalid message code", - ErrProtocolVersionMismatch: "Protocol version mismatch", - ErrNetworkIdMismatch: "NetworkId mismatch", - ErrGenesisBlockMismatch: "Genesis block mismatch", - ErrNoStatusMsg: "No status message", - ErrExtraStatusMsg: "Extra status message", - ErrSuspendedPeer: "Suspended peer", - ErrRequestRejected: "Request rejected", - ErrUnexpectedResponse: "Unexpected response", - ErrInvalidResponse: "Invalid response", - ErrTooManyTimeouts: "Too many request timeouts", - ErrMissingKey: "Key missing from list", -} - -type announceBlock struct { - Hash common.Hash // Hash of one particular block being announced - Number uint64 // Number of one particular block being announced - Td *big.Int // Total difficulty of one particular block being announced -} - -// announceData is the network packet for the block announcements. -type announceData struct { - Hash common.Hash // Hash of one particular block being announced - Number uint64 // Number of one particular block being announced - Td *big.Int // Total difficulty of one particular block being announced - ReorgDepth uint64 - Update keyValueList -} - -// sanityCheck verifies that the values are reasonable, as a DoS protection -func (a *announceData) sanityCheck() error { - if tdlen := a.Td.BitLen(); tdlen > 100 { - return fmt.Errorf("too large block TD: bitlen %d", tdlen) - } - return nil -} - -// sign adds a signature to the block announcement by the given privKey -func (a *announceData) sign(privKey *ecdsa.PrivateKey) { - rlp, _ := rlp.EncodeToBytes(announceBlock{a.Hash, a.Number, a.Td}) - sig, _ := crypto.Sign(crypto.Keccak256(rlp), privKey) - a.Update = a.Update.add("sign", sig) -} - -// checkSignature verifies if the block announcement has a valid signature by the given pubKey -func (a *announceData) checkSignature(id enode.ID, update keyValueMap) error { - var sig []byte - if err := update.get("sign", &sig); err != nil { - return err - } - rlp, _ := rlp.EncodeToBytes(announceBlock{a.Hash, a.Number, a.Td}) - recPubkey, err := crypto.SigToPub(crypto.Keccak256(rlp), sig) - if err != nil { - return err - } - if id == enode.PubkeyToIDV4(recPubkey) { - return nil - } - return errors.New("wrong signature") -} - -type blockInfo struct { - Hash common.Hash // Hash of one particular block being announced - Number uint64 // Number of one particular block being announced - Td *big.Int // Total difficulty of one particular block being announced -} - -// getBlockHeadersData represents a block header query. -type getBlockHeadersData struct { - Origin hashOrNumber // Block from which to retrieve headers - Amount uint64 // Maximum number of headers to retrieve - Skip uint64 // Blocks to skip between consecutive headers - Reverse bool // Query direction (false = rising towards latest, true = falling towards genesis) -} - -// hashOrNumber is a combined field for specifying an origin block. -type hashOrNumber struct { - Hash common.Hash // Block hash from which to retrieve headers (excludes Number) - Number uint64 // Block hash from which to retrieve headers (excludes Hash) -} - -// EncodeRLP is a specialized encoder for hashOrNumber to encode only one of the -// two contained union fields. -func (hn *hashOrNumber) EncodeRLP(w io.Writer) error { - if hn.Hash == (common.Hash{}) { - return rlp.Encode(w, hn.Number) - } - if hn.Number != 0 { - return fmt.Errorf("both origin hash (%x) and number (%d) provided", hn.Hash, hn.Number) - } - return rlp.Encode(w, hn.Hash) -} - -// DecodeRLP is a specialized decoder for hashOrNumber to decode the contents -// into either a block hash or a block number. -func (hn *hashOrNumber) DecodeRLP(s *rlp.Stream) error { - _, size, _ := s.Kind() - origin, err := s.Raw() - if err == nil { - switch { - case size == 32: - err = rlp.DecodeBytes(origin, &hn.Hash) - case size <= 8: - err = rlp.DecodeBytes(origin, &hn.Number) - default: - err = fmt.Errorf("invalid input size %d for origin", size) - } - } - return err -} - -// CodeData is the network response packet for a node data retrieval. -type CodeData []struct { - Value []byte -} diff --git a/les/protocol/keyvalueset.go b/les/protocol/keyvalueset.go new file mode 100644 index 0000000000..8beb5eca8c --- /dev/null +++ b/les/protocol/keyvalueset.go @@ -0,0 +1,86 @@ +// Copyright 2020 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package protocol + +import ( + "errors" + + "github.com/ethereum/go-ethereum/rlp" +) + +// ErrNonexistentEntry is returned if the specified key is non-existent. +var ErrNonexistentEntry = errors.New("the entry is non-existent") + +// KeyValueEntry is the entry contained in a List or Map +// which can be extended with no limitaion. +type KeyValueEntry struct { + Key string + Value rlp.RawValue +} + +// KeyValueList is a set of entries in list format. +// +// Usually KeyValueList is used as the container for +// protocol handshake. +type KeyValueList []KeyValueEntry + +// KeyValueMap is a set of entires in map format. +// All entires is identified with its key and saved +// in RLP-encoded format. +// +// Usually KeyValueMap is used as the container for +// protocol handshake. +type KeyValueMap map[string]rlp.RawValue + +// Add adds a new entry with specified key and value into list. +func (l KeyValueList) Add(key string, val interface{}) KeyValueList { + var entry KeyValueEntry + entry.Key = key + if val == nil { + val = uint64(0) // Use empty uint64 as default value + } + enc, err := rlp.EncodeToBytes(val) + if err == nil { + entry.Value = enc + } + return append(l, entry) +} + +// ToMap converts list format to map format. Also returns +// the total size of converted map. +func (l KeyValueList) ToMap() (KeyValueMap, uint64) { + m := make(KeyValueMap) + var size uint64 + for _, entry := range l { + m[entry.Key] = entry.Value + size += uint64(len(entry.Key)) + uint64(len(entry.Value)) + 8 + } + return m, size +} + +// Get retrieves contained entry with specified key, decode the +// retrieved data in the provided container(interface). +func (m KeyValueMap) Get(key string, val interface{}) error { + enc, ok := m[key] + if !ok { + return ErrNonexistentEntry + } + if val == nil { + return nil + } + return rlp.DecodeBytes(enc, val) +} diff --git a/les/protocol/keyvalueset_test.go b/les/protocol/keyvalueset_test.go new file mode 100644 index 0000000000..df900d8dba --- /dev/null +++ b/les/protocol/keyvalueset_test.go @@ -0,0 +1,47 @@ +// Copyright 2020 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package protocol + +import ( + "testing" + + "github.com/ethereum/go-ethereum/rlp" +) + +func TestKeyValueSet(t *testing.T) { + var cases = []struct { + key string + value interface{} + }{ + // {"key1", uint64(10)}, + {"key2", false}, + {"key3", nil}, + } + var list KeyValueList + for _, c := range cases { + list = list.Add(c.key, c.value) + } + blob, err := rlp.EncodeToBytes(list) + if err != nil { + t.Fatalf("Failed to encode keyvalue list: %v", err) + } + var dec KeyValueList + err = rlp.DecodeBytes(blob, &dec) + if err != nil { + t.Fatalf("Failed to decode keyvalue list: %v", err) + } +} diff --git a/les/protocol/protocol.go b/les/protocol/protocol.go new file mode 100644 index 0000000000..5ee3916b5b --- /dev/null +++ b/les/protocol/protocol.go @@ -0,0 +1,362 @@ +// Copyright 2016 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package protocol defines all protocol related structures which will be +// used in both server side and client side. +package protocol + +import ( + "crypto/ecdsa" + "errors" + "fmt" + "io" + "math/big" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/light" + "github.com/ethereum/go-ethereum/p2p/discv5" + "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/rlp" +) + +// Constants to match up protocol versions and messages +const ( + Lpv2 = 2 + Lpv3 = 3 +) + +// Supported versions of the les protocol (first is primary) +var ( + ClientProtocolVersions = []uint{Lpv2, Lpv3} + ServerProtocolVersions = []uint{Lpv2, Lpv3} + AdvertiseProtocolVersions = []uint{Lpv2} // clients are searching for the first advertised protocol in the list +) + +// Number of implemented message corresponding to different protocol versions. +var ProtocolLengths = map[uint]uint64{Lpv2: 22, Lpv3: 24} + +const ( + NetworkId = 1 // Default ethereum mainnet network ID + ProtocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message +) + +// les protocol message codes +const ( + // Protocol messages inherited from LPV1 + StatusMsg = 0x00 + AnnounceMsg = 0x01 + GetBlockHeadersMsg = 0x02 + BlockHeadersMsg = 0x03 + GetBlockBodiesMsg = 0x04 + BlockBodiesMsg = 0x05 + GetReceiptsMsg = 0x06 + ReceiptsMsg = 0x07 + GetCodeMsg = 0x0a + CodeMsg = 0x0b + + // Protocol messages introduced in LPV2 + GetProofsV2Msg = 0x0f + ProofsV2Msg = 0x10 + GetHelperTrieProofsMsg = 0x11 + HelperTrieProofsMsg = 0x12 + SendTxV2Msg = 0x13 + GetTxStatusMsg = 0x14 + TxStatusMsg = 0x15 + + // Protocol messages introduced in LPV3 + StopMsg = 0x16 + ResumeMsg = 0x17 +) + +// The maxmium amount of data requested per retrieval request. +const ( + MaxHeaderFetch = 192 // Amount of block headers to be fetched per retrieval request + MaxBodyFetch = 32 // Amount of block bodies to be fetched per retrieval request + MaxReceiptFetch = 128 // Amount of transaction receipts to allow fetching per request + MaxCodeFetch = 64 // Amount of contract codes to allow fetching per request + MaxProofsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request + MaxHelperTrieProofsFetch = 64 // Amount of helper tries to be fetched per retrieval request + MaxTxSend = 64 // Amount of transactions to be send per request + MaxTxStatus = 256 // Amount of transactions to queried per request +) + +type RequestInfo struct { + Name string + MaxCount uint64 +} + +var LesRequests = map[uint64]RequestInfo{ + GetBlockHeadersMsg: {"GetBlockHeaders", MaxHeaderFetch}, + GetBlockBodiesMsg: {"GetBlockBodies", MaxBodyFetch}, + GetReceiptsMsg: {"GetReceipts", MaxReceiptFetch}, + GetCodeMsg: {"GetCode", MaxCodeFetch}, + GetProofsV2Msg: {"GetProofsV2", MaxProofsFetch}, + GetHelperTrieProofsMsg: {"GetHelperTrieProofs", MaxHelperTrieProofsFetch}, + SendTxV2Msg: {"SendTxV2", MaxTxSend}, + GetTxStatusMsg: {"GetTxStatus", MaxTxStatus}, +} + +type ErrCode int + +const ( + ErrMsgTooLarge = iota + ErrDecode + ErrInvalidMsgCode + ErrProtocolVersionMismatch + ErrNetworkIdMismatch + ErrGenesisBlockMismatch + ErrNoStatusMsg + ErrExtraStatusMsg + ErrSuspendedPeer + ErrUselessPeer + ErrRequestRejected + ErrUnexpectedResponse + ErrInvalidResponse + ErrTooManyTimeouts + ErrMissingKey +) + +func (e ErrCode) String() string { + return errorToString[int(e)] +} + +// XXX change once legacy code is out +var errorToString = map[int]string{ + ErrMsgTooLarge: "Message too long", + ErrDecode: "Invalid message", + ErrInvalidMsgCode: "Invalid message code", + ErrProtocolVersionMismatch: "Protocol version mismatch", + ErrNetworkIdMismatch: "NetworkId mismatch", + ErrGenesisBlockMismatch: "Genesis block mismatch", + ErrNoStatusMsg: "No status message", + ErrExtraStatusMsg: "Extra status message", + ErrSuspendedPeer: "Suspended peer", + ErrRequestRejected: "Request rejected", + ErrUnexpectedResponse: "Unexpected response", + ErrInvalidResponse: "Invalid response", + ErrTooManyTimeouts: "Too many request timeouts", + ErrMissingKey: "Key missing from list", +} + +// HeadHeader is the a part of announcement sent by the LES server to the +// LES client when a new block is generated in the network. +// +// HeadHeader can also be used to represent the head info of peer(both server +// and client). +type HeadHeader struct { + Hash common.Hash // Hash of one particular block being announced + Number uint64 // Number of one particular block being announced + Td *big.Int // Total difficulty of one particular block being announced +} + +// Announcement is a network packet sent by the LES server to the LES client +// when a new block is generated in the network or the server has protocol +// parameters that need to be updated. +type Announcement struct { + HeadHeader // The data of new arrival header + ReorgDepth uint64 // The reorg depth of new arrival header + Update KeyValueList // Updated protocol parameters +} + +// SanityCheck verifies that the values are reasonable, as a DoS protection +func (a *Announcement) SanityCheck() error { + if tdlen := a.Td.BitLen(); tdlen > 100 { + return fmt.Errorf("too large block TD: bitlen %d", tdlen) + } + return nil +} + +// Sign adds a signature to the block announcement by the given privKey +func (a *Announcement) Sign(privKey *ecdsa.PrivateKey) { + rlp, _ := rlp.EncodeToBytes(HeadHeader{a.Hash, a.Number, a.Td}) + sig, _ := crypto.Sign(crypto.Keccak256(rlp), privKey) + a.Update = a.Update.Add("sign", sig) +} + +// CheckSignature verifies if the block announcement has a valid signature +// by the given pubKey. +func (a *Announcement) CheckSignature(id enode.ID, update KeyValueMap) error { + var sig []byte + if err := update.Get("sign", &sig); err != nil { + return err + } + rlp, _ := rlp.EncodeToBytes(HeadHeader{a.Hash, a.Number, a.Td}) + recPubkey, err := crypto.SigToPub(crypto.Keccak256(rlp), sig) + if err != nil { + return err + } + if id == enode.PubkeyToIDV4(recPubkey) { + return nil + } + return errors.New("wrong signature") +} + +// GetBlockHeadersRequest represents a block header query sent by les client. +type GetBlockHeadersRequest struct { + Origin HashOrNumber // Block from which to retrieve headers + Amount uint64 // Maximum number of headers to retrieve + Skip uint64 // Blocks to skip between consecutive headers + Reverse bool // Query direction (false = rising towards latest, true = falling towards genesis) +} + +// HashOrNumber is a combined field for specifying an origin block. +type HashOrNumber struct { + Hash common.Hash // Block hash from which to retrieve headers (excludes Number) + Number uint64 // Block hash from which to retrieve headers (excludes Hash) +} + +// EncodeRLP is a specialized encoder for HashOrNumber to encode only one of the +// two contained union fields. +func (hn *HashOrNumber) EncodeRLP(w io.Writer) error { + if hn.Hash == (common.Hash{}) { + return rlp.Encode(w, hn.Number) + } + if hn.Number != 0 { + return fmt.Errorf("both origin hash (%x) and number (%d) provided", hn.Hash, hn.Number) + } + return rlp.Encode(w, hn.Hash) +} + +// DecodeRLP is a specialized decoder for HashOrNumber to decode the contents +// into either a block hash or a block number. +func (hn *HashOrNumber) DecodeRLP(s *rlp.Stream) error { + _, size, _ := s.Kind() + origin, err := s.Raw() + if err == nil { + switch { + case size == 32: + err = rlp.DecodeBytes(origin, &hn.Hash) + case size <= 8: + err = rlp.DecodeBytes(origin, &hn.Number) + default: + err = fmt.Errorf("invalid input size %d for origin", size) + } + } + return err +} + +// TrieProofRequest represents a state/storage trie proof query +// sent by les client. +type TrieProofRequest struct { + BlockHash common.Hash // The corresponding block hash of state + Account []byte // The address of target account, nil if it's a global state trie proof request + Key []byte // The key of target storage slot or account + FromLevel uint // The node level beyond which all trie nodes are contained in the proof +} + +// CodeRequest represents a contract code query sent by les client. +type CodeRequest struct { + BlockHash common.Hash // The corresponding block hash of state + Account []byte // The address of target account +} + +const ( + // HelperTrieCHT is the indicator of canonical hash trie, check + // https://github.com/ethereum/devp2p/blob/master/caps/les.md#canonical-hash-trie + // for more details. + HelperTrieCHT = iota + + // HelperTrieBloomTrie is the indicator of bloom trie, check + // https://github.com/ethereum/devp2p/blob/master/caps/les.md#bloombits-trie + // for more details + HelperTrieBloomTrie + + // The auxiliary data type of helperTrie request which is available for + // all helperTrie request. + AuxRoot = 1 + + // The auxiliary data type of CHT request - corresponding block header + // which is only avaiable for CHT request. + AuxHeader = 2 +) + +// HelperTrieRequest represents a helper trie query sent by les client. +// HelperTrie includes: CHT and bloom trie. It's a shared structure between +// these two kinds of request. +// +// Except the helperTrie proof of requested entry will be returned, caller +// can specify more additional auxiliary data to be returned via `AuxType`. +type HelperTrieRequest struct { + Type uint // Indicator of request type, 0 represents CHT, 1 represents Bloom trie + TrieIndex uint64 // The index(section index) of requested trie + Key []byte // The list of entry keys, caller can request a batch of entries in a single request. + FromLevel uint // The node level beyond which all trie nodes are contained in the proof + AuxType uint // The type of auxiliary data requested +} + +// HelperTrieResponse represents the response of corresponding helperTrie +// request. A single response contains a batch of requested proofs and +// corresponding auxiliary data. +type HelperTrieResponse struct { + Proofs light.NodeList // The container for storing all requested proofs + AuxData [][]byte // The batch of requested auxiliary data +} + +// ErrResp returns an protocol error with given error code and additional +// error message. +func ErrResp(code ErrCode, format string, v ...interface{}) error { + return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...)) +} + +// LesTopic constructs the discovery v5 topic for LES protocol. +func LesTopic(genesisHash common.Hash, protocolVersion uint) discv5.Topic { + var name string + switch protocolVersion { + case Lpv2: + name = "LES2" + default: + panic(nil) + } + return discv5.Topic(name + "@" + common.Bytes2Hex(genesisHash.Bytes()[0:8])) +} + +type ( + // RequestCost represents a cost policy of a specified request type. + RequestCost struct { + BaseCost, ReqCost uint64 + } + // RequestCostTable assigns a cost estimate function to each request type + // which is a linear function of the requested amount + // (cost = BaseCost + ReqCost * amount) + RequestCostTable map[uint64]*RequestCost + // RequestCostList is a list representation of request costs which is used for + // database storage and communication through the network + RequestCostList []RequestCostListItem + RequestCostListItem struct { + MsgCode, BaseCost, ReqCost uint64 + } +) + +// GetMaxCost calculates the estimated cost for a given request type and amount +func (table RequestCostTable) GetMaxCost(code, amount uint64) uint64 { + costs := table[code] + return costs.BaseCost + amount*costs.ReqCost +} + +// ToTable converts a cost list to a cost table +func (list RequestCostList) ToTable(protocolLength uint64) RequestCostTable { + table := make(RequestCostTable) + for _, e := range list { + if e.MsgCode < protocolLength { + table[e.MsgCode] = &RequestCost{ + BaseCost: e.BaseCost, + ReqCost: e.ReqCost, + } + } + } + return table +} diff --git a/les/retrieve.go b/les/retrieve.go index c806117902..7d46ebf972 100644 --- a/les/retrieve.go +++ b/les/retrieve.go @@ -25,6 +25,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" ) @@ -171,7 +172,7 @@ func (rm *retrieveManager) deliver(peer distPeer, msg *Msg) error { if ok { return req.deliver(peer, msg) } - return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID) + return protocol.ErrResp(protocol.ErrUnexpectedResponse, "reqID = %v", msg.ReqID) } // frozen is called by the LES protocol manager when a server has suspended its service and we @@ -389,7 +390,7 @@ func (r *sentReq) deliver(peer distPeer, msg *Msg) error { s, ok := r.sentTo[peer] if !ok || s.delivered { - return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID) + return protocol.ErrResp(protocol.ErrUnexpectedResponse, "reqID = %v", msg.ReqID) } if s.frozen { return nil @@ -402,7 +403,7 @@ func (r *sentReq) deliver(peer distPeer, msg *Msg) error { s.event <- rpDeliveredInvalid } if !valid { - return errResp(ErrInvalidResponse, "reqID = %v", msg.ReqID) + return protocol.ErrResp(protocol.ErrInvalidResponse, "reqID = %v", msg.ReqID) } return nil } diff --git a/les/server.go b/les/server.go index a130cee2bb..6d9066f70e 100644 --- a/les/server.go +++ b/les/server.go @@ -26,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/les/checkpointoracle" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" @@ -58,9 +59,9 @@ type LesServer struct { func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) { // Collect les protocol version information supported by local node. - lesTopics := make([]discv5.Topic, len(AdvertiseProtocolVersions)) - for i, pv := range AdvertiseProtocolVersions { - lesTopics[i] = lesTopic(e.BlockChain().Genesis().Hash(), pv) + lesTopics := make([]discv5.Topic, len(protocol.AdvertiseProtocolVersions)) + for i, pv := range protocol.AdvertiseProtocolVersions { + lesTopics[i] = protocol.LesTopic(e.BlockChain().Genesis().Hash(), pv) } // Calculate the number of threads used to service the light client // requests based on the user-specified value. @@ -151,7 +152,7 @@ func (s *LesServer) APIs() []rpc.API { } func (s *LesServer) Protocols() []p2p.Protocol { - ps := s.makeProtocols(ServerProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} { + ps := s.makeProtocols(protocol.ServerProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} { if p := s.peers.Peer(peerIdToString(id)); p != nil { return p.Info() } diff --git a/les/server_handler.go b/les/server_handler.go index 4b505c2bc4..a34f0daa9a 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -31,6 +31,7 @@ import ( "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" @@ -43,15 +44,6 @@ const ( softResponseLimit = 2 * 1024 * 1024 // Target maximum size of returned blocks, headers or node data. estHeaderRlpSize = 500 // Approximate size of an RLP encoded block header ethVersion = 63 // equivalent eth version for the downloader - - MaxHeaderFetch = 192 // Amount of block headers to be fetched per retrieval request - MaxBodyFetch = 32 // Amount of block bodies to be fetched per retrieval request - MaxReceiptFetch = 128 // Amount of transaction receipts to allow fetching per request - MaxCodeFetch = 64 // Amount of contract codes to allow fetching per request - MaxProofsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request - MaxHelperTrieProofsFetch = 64 // Amount of helper tries to be fetched per retrieval request - MaxTxSend = 64 // Amount of transactions to be send per request - MaxTxStatus = 256 // Amount of transactions to queried per request ) var ( @@ -182,9 +174,9 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size) // Discard large message which exceeds the limitation. - if msg.Size > ProtocolMaxMsgSize { + if msg.Size > protocol.ProtocolMaxMsgSize { clientErrorMeter.Mark(1) - return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + return protocol.ErrResp(protocol.ErrMsgTooLarge, "%v > %v", msg.Size, protocol.ProtocolMaxMsgSize) } defer msg.Discard() @@ -204,7 +196,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { return false } // Prepaid max cost units before request been serving. - maxCost = p.fcCosts.getMaxCost(msg.Code, reqCnt) + maxCost = p.fcCosts.GetMaxCost(msg.Code, reqCnt) accepted, bufShort, priority := p.fcClient.AcceptRequest(reqID, responseCount, maxCost) if !accepted { p.freezeClient() @@ -268,7 +260,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } } switch msg.Code { - case GetBlockHeadersMsg: + case protocol.GetBlockHeadersMsg: p.Log().Trace("Received block header request") if metrics.EnabledExpensive { miscInHeaderPacketsMeter.Mark(1) @@ -277,14 +269,14 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } var req struct { ReqID uint64 - Query getBlockHeadersData + Query protocol.GetBlockHeadersRequest } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "%v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "%v: %v", msg, err) } query := req.Query - if accept(req.ReqID, query.Amount, MaxHeaderFetch) { + if accept(req.ReqID, query.Amount, protocol.MaxHeaderFetch) { wg.Add(1) go func() { defer wg.Done() @@ -381,7 +373,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { }() } - case GetBlockBodiesMsg: + case protocol.GetBlockBodiesMsg: p.Log().Trace("Received block bodies request") if metrics.EnabledExpensive { miscInBodyPacketsMeter.Mark(1) @@ -394,14 +386,14 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } var ( bytes int bodies []rlp.RawValue ) reqCnt := len(req.Hashes) - if accept(req.ReqID, uint64(reqCnt), MaxBodyFetch) { + if accept(req.ReqID, uint64(reqCnt), protocol.MaxBodyFetch) { wg.Add(1) go func() { defer wg.Done() @@ -430,7 +422,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { }() } - case GetCodeMsg: + case protocol.GetCodeMsg: p.Log().Trace("Received code request") if metrics.EnabledExpensive { miscInCodePacketsMeter.Mark(1) @@ -439,18 +431,18 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } var req struct { ReqID uint64 - Reqs []CodeReq + Reqs []protocol.CodeRequest } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } var ( bytes int data [][]byte ) reqCnt := len(req.Reqs) - if accept(req.ReqID, uint64(reqCnt), MaxCodeFetch) { + if accept(req.ReqID, uint64(reqCnt), protocol.MaxCodeFetch) { wg.Add(1) go func() { defer wg.Done() @@ -460,9 +452,9 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { return } // Look up the root hash belonging to the request - header := h.blockchain.GetHeaderByHash(request.BHash) + header := h.blockchain.GetHeaderByHash(request.BlockHash) if header == nil { - p.Log().Warn("Failed to retrieve associate header for code", "hash", request.BHash) + p.Log().Warn("Failed to retrieve associate header for code", "hash", request.BlockHash) atomic.AddUint32(&p.invalidCount, 1) continue } @@ -476,15 +468,15 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } triedb := h.blockchain.StateCache().TrieDB() - account, err := h.getAccount(triedb, header.Root, common.BytesToHash(request.AccKey)) + account, err := h.getAccount(triedb, header.Root, common.BytesToHash(request.Account)) if err != nil { - p.Log().Warn("Failed to retrieve account for code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "err", err) + p.Log().Warn("Failed to retrieve account for code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "err", err) atomic.AddUint32(&p.invalidCount, 1) continue } code, err := triedb.Node(common.BytesToHash(account.CodeHash)) if err != nil { - p.Log().Warn("Failed to retrieve account code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "codehash", common.BytesToHash(account.CodeHash), "err", err) + p.Log().Warn("Failed to retrieve account code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "codehash", common.BytesToHash(account.CodeHash), "err", err) continue } // Accumulate the code and abort if enough data was retrieved @@ -502,7 +494,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { }() } - case GetReceiptsMsg: + case protocol.GetReceiptsMsg: p.Log().Trace("Received receipts request") if metrics.EnabledExpensive { miscInReceiptPacketsMeter.Mark(1) @@ -515,14 +507,14 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } var ( bytes int receipts []rlp.RawValue ) reqCnt := len(req.Hashes) - if accept(req.ReqID, uint64(reqCnt), MaxReceiptFetch) { + if accept(req.ReqID, uint64(reqCnt), protocol.MaxReceiptFetch) { wg.Add(1) go func() { defer wg.Done() @@ -559,7 +551,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { }() } - case GetProofsV2Msg: + case protocol.GetProofsV2Msg: p.Log().Trace("Received les/2 proofs request") if metrics.EnabledExpensive { miscInTrieProofPacketsMeter.Mark(1) @@ -568,11 +560,11 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } var req struct { ReqID uint64 - Reqs []ProofReq + Reqs []protocol.TrieProofRequest } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } // Gather state data until the fetch or network limits is reached var ( @@ -580,7 +572,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { root common.Hash ) reqCnt := len(req.Reqs) - if accept(req.ReqID, uint64(reqCnt), MaxProofsFetch) { + if accept(req.ReqID, uint64(reqCnt), protocol.MaxProofsFetch) { wg.Add(1) go func() { defer wg.Done() @@ -596,11 +588,11 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { header *types.Header trie state.Trie ) - if request.BHash != lastBHash { - root, lastBHash = common.Hash{}, request.BHash + if request.BlockHash != lastBHash { + root, lastBHash = common.Hash{}, request.BlockHash - if header = h.blockchain.GetHeaderByHash(request.BHash); header == nil { - p.Log().Warn("Failed to retrieve header for proof", "hash", request.BHash) + if header = h.blockchain.GetHeaderByHash(request.BlockHash); header == nil { + p.Log().Warn("Failed to retrieve header for proof", "hash", request.BlockHash) atomic.AddUint32(&p.invalidCount, 1) continue } @@ -622,7 +614,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { // Open the account or storage trie for the request statedb := h.blockchain.StateCache() - switch len(request.AccKey) { + switch len(request.Account) { case 0: // No account key specified, open an account trie trie, err = statedb.OpenTrie(root) @@ -632,15 +624,15 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } default: // Account key specified, open a storage trie - account, err := h.getAccount(statedb.TrieDB(), root, common.BytesToHash(request.AccKey)) + account, err := h.getAccount(statedb.TrieDB(), root, common.BytesToHash(request.Account)) if err != nil { - p.Log().Warn("Failed to retrieve account for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "err", err) + p.Log().Warn("Failed to retrieve account for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "err", err) atomic.AddUint32(&p.invalidCount, 1) continue } - trie, err = statedb.OpenStorageTrie(common.BytesToHash(request.AccKey), account.Root) + trie, err = statedb.OpenStorageTrie(common.BytesToHash(request.Account), account.Root) if trie == nil || err != nil { - p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "root", account.Root, "err", err) + p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "root", account.Root, "err", err) continue } } @@ -662,7 +654,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { }() } - case GetHelperTrieProofsMsg: + case protocol.GetHelperTrieProofsMsg: p.Log().Trace("Received helper trie proof request") if metrics.EnabledExpensive { miscInHelperTriePacketsMeter.Mark(1) @@ -671,11 +663,11 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } var req struct { ReqID uint64 - Reqs []HelperTrieReq + Reqs []protocol.HelperTrieRequest } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } // Gather state data until the fetch or network limits is reached var ( @@ -683,7 +675,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { auxData [][]byte ) reqCnt := len(req.Reqs) - if accept(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) { + if accept(req.ReqID, uint64(reqCnt), protocol.MaxHelperTrieProofsFetch) { wg.Add(1) go func() { defer wg.Done() @@ -699,15 +691,15 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { sendResponse(req.ReqID, 0, nil, task.servingTime) return } - if auxTrie == nil || request.Type != lastType || request.TrieIdx != lastIdx { - auxTrie, lastType, lastIdx = nil, request.Type, request.TrieIdx + if auxTrie == nil || request.Type != lastType || request.TrieIndex != lastIdx { + auxTrie, lastType, lastIdx = nil, request.Type, request.TrieIndex var prefix string - if root, prefix = h.getHelperTrie(request.Type, request.TrieIdx); root != (common.Hash{}) { + if root, prefix = h.getHelperTrie(request.Type, request.TrieIndex); root != (common.Hash{}) { auxTrie, _ = trie.New(root, trie.NewDatabase(rawdb.NewTable(h.chainDb, prefix))) } } - if request.AuxReq == auxRoot { + if request.AuxType == protocol.AuxRoot { var data []byte if root != (common.Hash{}) { data = root[:] @@ -718,7 +710,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { if auxTrie != nil { auxTrie.Prove(request.Key, request.FromLevel, nodes) } - if request.AuxReq != 0 { + if request.AuxType != 0 { data := h.getAuxiliaryHeaders(request) auxData = append(auxData, data) auxBytes += len(data) @@ -728,7 +720,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { break } } - reply := p.ReplyHelperTrieProofs(req.ReqID, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData}) + reply := p.ReplyHelperTrieProofs(req.ReqID, protocol.HelperTrieResponse{Proofs: nodes.NodeList(), AuxData: auxData}) sendResponse(req.ReqID, uint64(reqCnt), reply, task.done()) if metrics.EnabledExpensive { miscOutHelperTriePacketsMeter.Mark(1) @@ -737,7 +729,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { }() } - case SendTxV2Msg: + case protocol.SendTxV2Msg: p.Log().Trace("Received new transactions") if metrics.EnabledExpensive { miscInTxsPacketsMeter.Mark(1) @@ -750,10 +742,10 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } reqCnt := len(req.Txs) - if accept(req.ReqID, uint64(reqCnt), MaxTxSend) { + if accept(req.ReqID, uint64(reqCnt), protocol.MaxTxSend) { wg.Add(1) go func() { defer wg.Done() @@ -786,7 +778,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { }() } - case GetTxStatusMsg: + case protocol.GetTxStatusMsg: p.Log().Trace("Received transaction status query request") if metrics.EnabledExpensive { miscInTxStatusPacketsMeter.Mark(1) @@ -799,10 +791,10 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { } if err := msg.Decode(&req); err != nil { clientErrorMeter.Mark(1) - return errResp(ErrDecode, "msg %v: %v", msg, err) + return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err) } reqCnt := len(req.Hashes) - if accept(req.ReqID, uint64(reqCnt), MaxTxStatus) { + if accept(req.ReqID, uint64(reqCnt), protocol.MaxTxStatus) { wg.Add(1) go func() { defer wg.Done() @@ -826,7 +818,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { default: p.Log().Trace("Received invalid message", "code", msg.Code) clientErrorMeter.Mark(1) - return errResp(ErrInvalidMsgCode, "%v", msg.Code) + return protocol.ErrResp(protocol.ErrInvalidMsgCode, "%v", msg.Code) } // If the client has made too much invalid request(e.g. request a non-exist data), // reject them to prevent SPAM attack. @@ -857,10 +849,10 @@ func (h *serverHandler) getAccount(triedb *trie.Database, root, hash common.Hash // getHelperTrie returns the post-processed trie root for the given trie ID and section index func (h *serverHandler) getHelperTrie(typ uint, index uint64) (common.Hash, string) { switch typ { - case htCanonical: + case protocol.HelperTrieCHT: sectionHead := rawdb.ReadCanonicalHash(h.chainDb, (index+1)*h.server.iConfig.ChtSize-1) return light.GetChtRoot(h.chainDb, index, sectionHead), light.ChtTablePrefix - case htBloomBits: + case protocol.HelperTrieBloomTrie: sectionHead := rawdb.ReadCanonicalHash(h.chainDb, (index+1)*h.server.iConfig.BloomTrieSize-1) return light.GetBloomTrieRoot(h.chainDb, index, sectionHead), light.BloomTrieTablePrefix } @@ -868,8 +860,8 @@ func (h *serverHandler) getHelperTrie(typ uint, index uint64) (common.Hash, stri } // getAuxiliaryHeaders returns requested auxiliary headers for the CHT request. -func (h *serverHandler) getAuxiliaryHeaders(req HelperTrieReq) []byte { - if req.Type == htCanonical && req.AuxReq == auxHeader && len(req.Key) == 8 { +func (h *serverHandler) getAuxiliaryHeaders(req protocol.HelperTrieRequest) []byte { + if req.Type == protocol.HelperTrieCHT && req.AuxType == protocol.AuxHeader && len(req.Key) == 8 { blockNum := binary.BigEndian.Uint64(req.Key) hash := rawdb.ReadCanonicalHash(h.chainDb, blockNum) return rawdb.ReadHeaderRLP(h.chainDb, hash, blockNum) @@ -931,9 +923,16 @@ func (h *serverHandler) broadcastHeaders() { log.Debug("Announcing block to peers", "number", number, "hash", hash, "td", td, "reorg", reorg) var ( signed bool - signedAnnounce announceData + signedAnnounce protocol.Announcement ) - announce := announceData{Hash: hash, Number: number, Td: td, ReorgDepth: reorg} + announce := protocol.Announcement{ + HeadHeader: protocol.HeadHeader{ + Hash: hash, + Number: number, + Td: td, + }, + ReorgDepth: reorg, + } for _, p := range peers { p := p switch p.announceType { @@ -942,7 +941,7 @@ func (h *serverHandler) broadcastHeaders() { case announceTypeSigned: if !signed { signedAnnounce = announce - signedAnnounce.sign(h.server.privateKey) + signedAnnounce.Sign(h.server.privateKey) signed = true } p.queueSend(func() { p.SendAnnounce(signedAnnounce) }) diff --git a/les/test_helper.go b/les/test_helper.go index 5c69050e22..47959ca5be 100644 --- a/les/test_helper.go +++ b/les/test_helper.go @@ -41,6 +41,7 @@ import ( "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/les/checkpointoracle" "github.com/ethereum/go-ethereum/les/flowcontrol" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" @@ -200,7 +201,7 @@ func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, index client := &LightEthereum{ lesCommons: lesCommons{ genesis: genesis.Hash(), - config: ð.Config{LightPeers: 100, NetworkId: NetworkId}, + config: ð.Config{LightPeers: 100, NetworkId: protocol.NetworkId}, chainConfig: params.AllEthashProtocolChanges, iConfig: light.TestClientIndexerConfig, chainDb: db, @@ -263,7 +264,7 @@ func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Da server := &LesServer{ lesCommons: lesCommons{ genesis: genesis.Hash(), - config: ð.Config{LightPeers: 100, NetworkId: NetworkId}, + config: ð.Config{LightPeers: 100, NetworkId: protocol.NetworkId}, chainConfig: params.AllEthashProtocolChanges, iConfig: light.TestServerIndexerConfig, chainDb: db, @@ -308,7 +309,7 @@ func newTestPeer(t *testing.T, name string, version int, handler *serverHandler, // Generate a random id and create the peer var id enode.ID rand.Read(id[:]) - peer := newPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), net) + peer := newPeer(version, protocol.NetworkId, false, p2p.NewPeer(id, name, nil), net) // Start the peer on a new thread errCh := make(chan error, 1) @@ -354,8 +355,8 @@ func newTestPeerPair(name string, version int, server *serverHandler, client *cl var id enode.ID rand.Read(id[:]) - peer1 := newPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), net) - peer2 := newPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), app) + peer1 := newPeer(version, protocol.NetworkId, false, p2p.NewPeer(id, name, nil), net) + peer2 := newPeer(version, protocol.NetworkId, false, p2p.NewPeer(id, name, nil), app) // Start the peer on a new thread errc1 := make(chan error, 1) @@ -379,29 +380,29 @@ func newTestPeerPair(name string, version int, server *serverHandler, client *cl // handshake simulates a trivial handshake that expects the same state from the // remote side as we are simulating locally. -func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, costList RequestCostList) { - var expList keyValueList - expList = expList.add("protocolVersion", uint64(p.peer.version)) - expList = expList.add("networkId", uint64(NetworkId)) - expList = expList.add("headTd", td) - expList = expList.add("headHash", head) - expList = expList.add("headNum", headNum) - expList = expList.add("genesisHash", genesis) - sendList := make(keyValueList, len(expList)) +func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, costList protocol.RequestCostList) { + var expList protocol.KeyValueList + expList = expList.Add("protocolVersion", uint64(p.peer.version)) + expList = expList.Add("networkId", uint64(protocol.NetworkId)) + expList = expList.Add("headTd", td) + expList = expList.Add("headHash", head) + expList = expList.Add("headNum", headNum) + expList = expList.Add("genesisHash", genesis) + sendList := make(protocol.KeyValueList, len(expList)) copy(sendList, expList) - expList = expList.add("serveHeaders", nil) - expList = expList.add("serveChainSince", uint64(0)) - expList = expList.add("serveStateSince", uint64(0)) - expList = expList.add("serveRecentState", uint64(core.TriesInMemory-4)) - expList = expList.add("txRelay", nil) - expList = expList.add("flowControl/BL", testBufLimit) - expList = expList.add("flowControl/MRR", testBufRecharge) - expList = expList.add("flowControl/MRC", costList) + expList = expList.Add("serveHeaders", nil) + expList = expList.Add("serveChainSince", uint64(0)) + expList = expList.Add("serveStateSince", uint64(0)) + expList = expList.Add("serveRecentState", uint64(core.TriesInMemory-4)) + expList = expList.Add("txRelay", nil) + expList = expList.Add("flowControl/BL", testBufLimit) + expList = expList.Add("flowControl/MRR", testBufRecharge) + expList = expList.Add("flowControl/MRC", costList) - if err := p2p.ExpectMsg(p.app, StatusMsg, expList); err != nil { + if err := p2p.ExpectMsg(p.app, protocol.StatusMsg, expList); err != nil { t.Fatalf("status recv: %v", err) } - if err := p2p.Send(p.app, StatusMsg, sendList); err != nil { + if err := p2p.Send(p.app, protocol.StatusMsg, sendList); err != nil { t.Fatalf("status send: %v", err) } p.peer.fcParams = flowcontrol.ServerParams{ diff --git a/les/ulc_test.go b/les/ulc_test.go index 9112bf928c..6bfed394e9 100644 --- a/les/ulc_test.go +++ b/les/ulc_test.go @@ -24,6 +24,7 @@ import ( "time" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/les/protocol" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" ) @@ -86,15 +87,15 @@ func testULCAnnounceThreshold(t *testing.T, protocol int) { } } -func connect(server *serverHandler, serverId enode.ID, client *clientHandler, protocol int) (*peer, *peer, error) { +func connect(server *serverHandler, serverId enode.ID, client *clientHandler, p int) (*peer, *peer, error) { // Create a message pipe to communicate through app, net := p2p.MsgPipe() var id enode.ID rand.Read(id[:]) - peer1 := newPeer(protocol, NetworkId, true, p2p.NewPeer(serverId, "", nil), net) // Mark server as trusted - peer2 := newPeer(protocol, NetworkId, false, p2p.NewPeer(id, "", nil), app) + peer1 := newPeer(p, protocol.NetworkId, true, p2p.NewPeer(serverId, "", nil), net) // Mark server as trusted + peer2 := newPeer(p, protocol.NetworkId, false, p2p.NewPeer(id, "", nil), app) // Start the peerLight on a new thread errc1 := make(chan error, 1)