forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

les: add checkpoint challenge

Browse source
This commit is contained in:
rjl493456442 2019-09-25 15:07:58 +08:00
parent 52a8f7dfd3
commit f0869777e7
8 changed files with 175 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
les/benchmark.go
View file

@ -74,9 +74,9 @@ func (b *benchmarkBlockHeaders) init(h *serverHandler, count int) error {
func (b *benchmarkBlockHeaders) request(peer *peer, index int) error { func (b *benchmarkBlockHeaders) request(peer *peer, index int) error {
if b.byHash { if b.byHash {
return peer.RequestHeadersByHash(0, 0, b.hashes[index], b.amount, b.skip, b.reverse) return peer.RequestHeadersByHash(0, b.hashes[index], b.amount, b.skip, b.reverse)
} else { } else {
return peer.RequestHeadersByNumber(0, 0, uint64(b.offset+rand.Int63n(b.randMax)), b.amount, b.skip, b.reverse) return peer.RequestHeadersByNumber(0, uint64(b.offset+rand.Int63n(b.randMax)), b.amount, b.skip, b.reverse)
} }
} }
@ -97,9 +97,9 @@ func (b *benchmarkBodiesOrReceipts) init(h *serverHandler, count int) error {
func (b *benchmarkBodiesOrReceipts) request(peer *peer, index int) error { func (b *benchmarkBodiesOrReceipts) request(peer *peer, index int) error {
if b.receipts { if b.receipts {
return peer.RequestReceipts(0, 0, []common.Hash{b.hashes[index]}) return peer.RequestReceipts(0, []common.Hash{b.hashes[index]})
} else { } else {
return peer.RequestBodies(0, 0, []common.Hash{b.hashes[index]}) return peer.RequestBodies(0, []common.Hash{b.hashes[index]})
} }
} }
@ -118,9 +118,9 @@ func (b *benchmarkProofsOrCode) request(peer *peer, index int) error {
key := make([]byte, 32) key := make([]byte, 32)
rand.Read(key) rand.Read(key)
if b.code { if b.code {
return peer.RequestCode(0, 0, []CodeReq{{BHash: b.headHash, AccKey: key}}) return peer.RequestCode(0, []CodeReq{{BHash: b.headHash, AccKey: key}})
} else { } else {
return peer.RequestProofs(0, 0, []ProofReq{{BHash: b.headHash, Key: key}}) return peer.RequestProofs(0, []ProofReq{{BHash: b.headHash, Key: key}})
} }
} }
@ -163,7 +163,7 @@ func (b *benchmarkHelperTrie) request(peer *peer, index int) error {
} }
} }
return peer.RequestHelperTrieProofs(0, 0, reqs) return peer.RequestHelperTrieProofs(0, reqs)
} }
// benchmarkTxSend implements requestBenchmark // benchmarkTxSend implements requestBenchmark
@ -191,7 +191,7 @@ func (b *benchmarkTxSend) init(h *serverHandler, count int) error {
func (b *benchmarkTxSend) request(peer *peer, index int) error { func (b *benchmarkTxSend) request(peer *peer, index int) error {
enc, _ := rlp.EncodeToBytes(types.Transactions{b.txs[index]}) enc, _ := rlp.EncodeToBytes(types.Transactions{b.txs[index]})
return peer.SendTxs(0, 0, enc) return peer.SendTxs(0, enc)
} }
// benchmarkTxStatus implements requestBenchmark // benchmarkTxStatus implements requestBenchmark
@ -204,7 +204,7 @@ func (b *benchmarkTxStatus) init(h *serverHandler, count int) error {
func (b *benchmarkTxStatus) request(peer *peer, index int) error { func (b *benchmarkTxStatus) request(peer *peer, index int) error {
var hash common.Hash var hash common.Hash
rand.Read(hash[:]) rand.Read(hash[:])
return peer.RequestTxStatus(0, 0, []common.Hash{hash}) return peer.RequestTxStatus(0, []common.Hash{hash})
} }
// benchmarkSetup stores measurement data for a single benchmark type // benchmarkSetup stores measurement data for a single benchmark type

85
les/client_handler.go
View file

@ -31,23 +31,30 @@ import (
"github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/params"
) )
var checkpointChallengeTimeout = 15 * time.Second // Time allowance for a node to reply to the checkpoint progress challenge
// clientHandler is responsible for receiving and processing all incoming server // clientHandler is responsible for receiving and processing all incoming server
// responses. // responses.
type clientHandler struct { type clientHandler struct {
ulc *ulc ulc *ulc
clock mclock.Clock
checkpoint *params.TrustedCheckpoint checkpoint *params.TrustedCheckpoint
fetcher *lightFetcher fetcher *lightFetcher
downloader *downloader.Downloader downloader *downloader.Downloader
backend *LightEthereum backend *LightEthereum
closeCh chan struct{} closeCh chan struct{}
wg sync.WaitGroup // WaitGroup used to track all connected peers. wg sync.WaitGroup // WaitGroup used to track all connected peers.
syncDone func() // Test hooks when syncing is done.
// Testing fields or hooks
ignoreHeaders bool // Indicator whether ignore received headers
syncDone func() // Test hooks when syncing is done.
} }
func newClientHandler(ulcServers []string, ulcFraction int, checkpoint *params.TrustedCheckpoint, backend *LightEthereum) *clientHandler { func newClientHandler(ulcServers []string, ulcFraction int, checkpoint *params.TrustedCheckpoint, backend *LightEthereum) *clientHandler {
handler := &clientHandler{ handler := &clientHandler{
checkpoint: checkpoint, checkpoint: checkpoint,
clock: mclock.System{},
backend: backend, backend: backend,
closeCh: make(chan struct{}), closeCh: make(chan struct{}),
} }
@ -89,6 +96,7 @@ func (h *clientHandler) runPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter)
} }
h.wg.Add(1) h.wg.Add(1)
defer h.wg.Done() defer h.wg.Done()
err := h.handle(peer) err := h.handle(peer)
h.backend.serverPool.disconnect(peer.poolEntry) h.backend.serverPool.disconnect(peer.poolEntry)
return err return err
@ -131,6 +139,26 @@ func (h *clientHandler) handle(p *peer) error {
if p.poolEntry != nil { if p.poolEntry != nil {
h.backend.serverPool.registered(p.poolEntry) h.backend.serverPool.registered(p.poolEntry)
} }
// If we have a trusted CHT, reject all peers below that (avoid light sync eclipse)
if h.checkpoint != nil {
// Request the peer's checkpoint header for chain height/weight validation
wrapPeer := &peerConnection{handler: h, peer: p}
if err := wrapPeer.RequestHeadersByNumber((h.checkpoint.SectionIndex+1)*h.backend.iConfig.ChtSize-1, 1, 0, false); err != nil {
return err
}
// Start a timer to disconnect if the peer doesn't reply in time
p.syncDrop = h.clock.AfterFunc(checkpointChallengeTimeout, func() {
p.Log().Warn("Checkpoint challenge timed out, dropping", "addr", p.RemoteAddr(), "type", p.Name())
h.removePeer(p.id)
})
// Make sure it's cleaned up if the peer dies off
defer func() {
if p.syncDrop != nil {
p.syncDrop.Stop()
p.syncDrop = nil
}
}()
}
// Spawn a main loop to handle all incoming messages. // Spawn a main loop to handle all incoming messages.
for { for {
if err := h.handleMsg(p); err != nil { if err := h.handleMsg(p); err != nil {
@ -199,6 +227,29 @@ func (h *clientHandler) handleMsg(p *peer) error {
return errResp(ErrDecode, "msg %v: %v", msg, err) return errResp(ErrDecode, "msg %v: %v", msg, err)
} }
p.fcServer.ReceivedReply(resp.ReqID, resp.BV) p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
// If we are still waiting the checkpoint response.
if !h.ignoreHeaders && p.syncDrop != nil {
// First stop timer anyway.
p.syncDrop.Stop()
p.syncDrop = nil
// If no headers were received or more headers received than we expect,
// reject the server directly.
//
// Two cases here:
// (1) The server is not synced, so no checkpoint header to response
// (2) The server sends us useless headers which we don't explicitly
// request.
if len(resp.Headers) != 1 {
return errResp(ErrUselessPeer, "msg %v: %v", msg, err)
} else {
header := resp.Headers[0]
if header.Hash() != h.checkpoint.SectionHead {
return errResp(ErrUselessPeer, "msg %v: %v", msg, err)
}
}
}
// Deliver response header to concrete requester.
if h.fetcher.requestedID(resp.ReqID) { if h.fetcher.requestedID(resp.ReqID) {
h.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers) h.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers)
} else { } else {
@ -339,19 +390,13 @@ func (pc *peerConnection) Head() (common.Hash, *big.Int) {
func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error { func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error {
rq := &distReq{ rq := &distReq{
getCost: func(dp distPeer) uint64 { getCost: func(dp distPeer) uint64 { return dp.(*peer).GetRequestCost(GetBlockHeadersMsg, amount) },
peer := dp.(*peer) canSend: func(dp distPeer) bool { return dp.(*peer) == pc.peer },
return peer.GetRequestCost(GetBlockHeadersMsg, amount)
},
canSend: func(dp distPeer) bool {
return dp.(*peer) == pc.peer
},
request: func(dp distPeer) func() { request: func(dp distPeer) func() {
reqID := genReqID() reqID := genReqID()
peer := dp.(*peer) peer := dp.(*peer)
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) peer.fcServer.QueuedRequest(reqID, peer.GetRequestCost(GetBlockHeadersMsg, amount))
peer.fcServer.QueuedRequest(reqID, cost) return func() { peer.RequestHeadersByHash(reqID, origin, amount, skip, reverse) }
return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) }
}, },
} }
_, ok := <-pc.handler.backend.reqDist.queue(rq) _, ok := <-pc.handler.backend.reqDist.queue(rq)
@ -363,19 +408,13 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s
func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error { func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error {
rq := &distReq{ rq := &distReq{
getCost: func(dp distPeer) uint64 { getCost: func(dp distPeer) uint64 { return dp.(*peer).GetRequestCost(GetBlockHeadersMsg, amount) },
peer := dp.(*peer) canSend: func(dp distPeer) bool { return dp.(*peer) == pc.peer },
return peer.GetRequestCost(GetBlockHeadersMsg, amount)
},
canSend: func(dp distPeer) bool {
return dp.(*peer) == pc.peer
},
request: func(dp distPeer) func() { request: func(dp distPeer) func() {
reqID := genReqID() reqID := genReqID()
peer := dp.(*peer) peer := dp.(*peer)
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) peer.fcServer.QueuedRequest(reqID, peer.GetRequestCost(GetBlockHeadersMsg, amount))
peer.fcServer.QueuedRequest(reqID, cost) return func() { peer.RequestHeadersByNumber(reqID, origin, amount, skip, reverse) }
return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) }
}, },
} }
_, ok := <-pc.handler.backend.reqDist.queue(rq) _, ok := <-pc.handler.backend.reqDist.queue(rq)

2
les/fetcher.go
View file

@ -557,7 +557,7 @@ func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bes
time.Sleep(hardRequestTimeout) time.Sleep(hardRequestTimeout)
f.timeoutChn <- reqID f.timeoutChn <- reqID
}() }()
return func() { p.RequestHeadersByHash(reqID, cost, bestHash, int(bestAmount), 0, true) } return func() { p.RequestHeadersByHash(reqID, bestHash, int(bestAmount), 0, true) }
}, },
} }
} }

97
les/handler_test.go
View file

@ -168,8 +168,7 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
// Send the hash request and verify the response // Send the hash request and verify the response
reqID++ reqID++
cost := server.peer.peer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount)) sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, tt.query)
sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, cost, tt.query)
if err := expectResponse(server.peer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil { if err := expectResponse(server.peer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil {
t.Errorf("test %d: headers mismatch: %v", i, err) t.Errorf("test %d: headers mismatch: %v", i, err)
} }
@ -246,8 +245,7 @@ func testGetBlockBodies(t *testing.T, protocol int) {
reqID++ reqID++
// Send the hash request and verify the response // Send the hash request and verify the response
cost := server.peer.peer.GetRequestCost(GetBlockBodiesMsg, len(hashes)) sendRequest(server.peer.app, GetBlockBodiesMsg, reqID, hashes)
sendRequest(server.peer.app, GetBlockBodiesMsg, reqID, cost, hashes)
if err := expectResponse(server.peer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil { if err := expectResponse(server.peer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil {
t.Errorf("test %d: bodies mismatch: %v", i, err) t.Errorf("test %d: bodies mismatch: %v", i, err)
} }
@ -278,8 +276,7 @@ func testGetCode(t *testing.T, protocol int) {
} }
} }
cost := server.peer.peer.GetRequestCost(GetCodeMsg, len(codereqs)) sendRequest(server.peer.app, GetCodeMsg, 42, codereqs)
sendRequest(server.peer.app, GetCodeMsg, 42, cost, codereqs)
if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, codes); err != nil { if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, codes); err != nil {
t.Errorf("codes mismatch: %v", err) t.Errorf("codes mismatch: %v", err)
} }
@ -299,8 +296,7 @@ func testGetStaleCode(t *testing.T, protocol int) {
BHash: bc.GetHeaderByNumber(number).Hash(), BHash: bc.GetHeaderByNumber(number).Hash(),
AccKey: crypto.Keccak256(testContractAddr[:]), AccKey: crypto.Keccak256(testContractAddr[:]),
} }
cost := server.peer.peer.GetRequestCost(GetCodeMsg, 1) sendRequest(server.peer.app, GetCodeMsg, 42, []*CodeReq{req})
sendRequest(server.peer.app, GetCodeMsg, 42, cost, []*CodeReq{req})
if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, expected); err != nil { if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, expected); err != nil {
t.Errorf("codes mismatch: %v", err) t.Errorf("codes mismatch: %v", err)
} }
@ -331,8 +327,7 @@ func testGetReceipt(t *testing.T, protocol int) {
receipts = append(receipts, rawdb.ReadRawReceipts(server.db, block.Hash(), block.NumberU64())) receipts = append(receipts, rawdb.ReadRawReceipts(server.db, block.Hash(), block.NumberU64()))
} }
// Send the hash request and verify the response // Send the hash request and verify the response
cost := server.peer.peer.GetRequestCost(GetReceiptsMsg, len(hashes)) sendRequest(server.peer.app, GetReceiptsMsg, 42, hashes)
sendRequest(server.peer.app, GetReceiptsMsg, 42, cost, hashes)
if err := expectResponse(server.peer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil { if err := expectResponse(server.peer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil {
t.Errorf("receipts mismatch: %v", err) t.Errorf("receipts mismatch: %v", err)
} }
@ -367,8 +362,7 @@ func testGetProofs(t *testing.T, protocol int) {
} }
} }
// Send the proof request and verify the response // Send the proof request and verify the response
cost := server.peer.peer.GetRequestCost(GetProofsV2Msg, len(proofreqs)) sendRequest(server.peer.app, GetProofsV2Msg, 42, proofreqs)
sendRequest(server.peer.app, GetProofsV2Msg, 42, cost, proofreqs)
if err := expectResponse(server.peer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil { if err := expectResponse(server.peer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil {
t.Errorf("proofs mismatch: %v", err) t.Errorf("proofs mismatch: %v", err)
} }
@ -392,8 +386,7 @@ func testGetStaleProof(t *testing.T, protocol int) {
BHash: header.Hash(), BHash: header.Hash(),
Key: account, Key: account,
} }
cost := server.peer.peer.GetRequestCost(GetProofsV2Msg, 1) sendRequest(server.peer.app, GetProofsV2Msg, 42, []*ProofReq{req})
sendRequest(server.peer.app, GetProofsV2Msg, 42, cost, []*ProofReq{req})
var expected []rlp.RawValue var expected []rlp.RawValue
if wantOK { if wantOK {
@ -453,8 +446,7 @@ func testGetCHTProofs(t *testing.T, protocol int) {
AuxReq: auxHeader, AuxReq: auxHeader,
}} }}
// Send the proof request and verify the response // Send the proof request and verify the response
cost := server.peer.peer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2)) sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, requestsV2)
sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2)
if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil { if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil {
t.Errorf("proofs mismatch: %v", err) t.Errorf("proofs mismatch: %v", err)
} }
@ -502,8 +494,7 @@ func testGetBloombitsProofs(t *testing.T, protocol int) {
trie.Prove(key, 0, &proofs.Proofs) trie.Prove(key, 0, &proofs.Proofs)
// Send the proof request and verify the response // Send the proof request and verify the response
cost := server.peer.peer.GetRequestCost(GetHelperTrieProofsMsg, len(requests)) sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, requests)
sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, cost, requests)
if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil { if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil {
t.Errorf("bit %d: proofs mismatch: %v", bit, err) t.Errorf("bit %d: proofs mismatch: %v", bit, err)
} }
@ -525,11 +516,9 @@ func testTransactionStatus(t *testing.T, protocol int) {
test := func(tx *types.Transaction, send bool, expStatus light.TxStatus) { test := func(tx *types.Transaction, send bool, expStatus light.TxStatus) {
reqID++ reqID++
if send { if send {
cost := server.peer.peer.GetRequestCost(SendTxV2Msg, 1) sendRequest(server.peer.app, SendTxV2Msg, reqID, types.Transactions{tx})
sendRequest(server.peer.app, SendTxV2Msg, reqID, cost, types.Transactions{tx})
} else { } else {
cost := server.peer.peer.GetRequestCost(GetTxStatusMsg, 1) sendRequest(server.peer.app, GetTxStatusMsg, reqID, []common.Hash{tx.Hash()})
sendRequest(server.peer.app, 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, TxStatusMsg, reqID, testBufLimit, []light.TxStatus{expStatus}); err != nil {
t.Errorf("transaction status mismatch") t.Errorf("transaction status mismatch")
@ -620,7 +609,7 @@ func TestStopResumeLes3(t *testing.T) {
header := server.handler.blockchain.CurrentHeader() header := server.handler.blockchain.CurrentHeader()
req := func() { req := func() {
reqID++ reqID++
sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, testCost, &getBlockHeadersData{Origin: hashOrNumber{Hash: header.Hash()}, Amount: 1}) sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, &getBlockHeadersData{Origin: hashOrNumber{Hash: header.Hash()}, Amount: 1})
} }
for i := 1; i <= 5; i++ { for i := 1; i <= 5; i++ {
// send requests while we still have enough buffer and expect a response // send requests while we still have enough buffer and expect a response
@ -651,3 +640,65 @@ func TestStopResumeLes3(t *testing.T) {
} }
} }
} }
func TestCheckpointChallengeLes2(t *testing.T) { testCheckpointChallenge(t, 2) }
func TestCheckpointChallengeLes3(t *testing.T) { testCheckpointChallenge(t, 3) }
func testCheckpointChallenge(t *testing.T, protocol int) {
config := light.TestServerIndexerConfig
waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
for {
cs, _, _ := cIndexer.Sections()
bts, _, _ := btIndexer.Sections()
if cs >= 1 && bts >= 1 {
break
}
time.Sleep(10 * time.Millisecond)
}
}
// Generate 512+4 blocks (totally 1 CHT sections)
server, client, tearDown := newClientServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, nil, 0, false, false)
defer tearDown()
s, _, head := server.chtIndexer.Sections()
cp := &params.TrustedCheckpoint{
SectionIndex: 0,
SectionHead: head,
CHTRoot: light.GetChtRoot(server.db, s-1, head),
BloomRoot: light.GetBloomTrieRoot(server.db, s-1, head),
}
// Register the assembled checkpoint as hardcoded one.
client.handler.clock = &mclock.Simulated{}
client.handler.checkpoint = cp
client.handler.backend.blockchain.AddTrustedCheckpoint(cp)
// Create connected peer pair.
_, err1, _, err2 := newTestPeerPair("peer", protocol, server.handler, client.handler)
select {
case <-time.After(time.Millisecond * 100):
case err := <-err1:
t.Fatalf("peer 1 handshake error: %v", err)
case err := <-err2:
t.Fatalf("peer 2 handshake error: %v", err)
}
client.handler.clock.(*mclock.Simulated).Run(checkpointChallengeTimeout)
if client.handler.backend.peers.Len() != 1 {
t.Fatalf("Should pass checkpoint challenge")
}
client.handler.ignoreHeaders = true // Explicitly ignore all received headers, trigger timer
// Create connected peer pair.
_, err1, _, err2 = newTestPeerPair("peer2", protocol, server.handler, client.handler)
select {
case <-time.After(time.Millisecond * 100):
case err := <-err1:
t.Fatalf("peer 1 handshake error: %v", err)
case err := <-err2:
t.Fatalf("peer 2 handshake error: %v", err)
}
client.handler.clock.(*mclock.Simulated).Run(checkpointChallengeTimeout)
if client.handler.backend.peers.Len() != 1 {
t.Fatalf("Shouldn't pass checkpoint challenge")
}
}

14
les/odr_requests.go
View file

@ -90,7 +90,7 @@ func (r *BlockRequest) CanSend(peer *peer) bool {
// Request sends an ODR request to the LES network (implementation of LesOdrRequest) // Request sends an ODR request to the LES network (implementation of LesOdrRequest)
func (r *BlockRequest) Request(reqID uint64, peer *peer) error { func (r *BlockRequest) Request(reqID uint64, peer *peer) error {
peer.Log().Debug("Requesting block body", "hash", r.Hash) peer.Log().Debug("Requesting block body", "hash", r.Hash)
return peer.RequestBodies(reqID, r.GetCost(peer), []common.Hash{r.Hash}) return peer.RequestBodies(reqID, []common.Hash{r.Hash})
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network
@ -146,7 +146,7 @@ func (r *ReceiptsRequest) CanSend(peer *peer) bool {
// Request sends an ODR request to the LES network (implementation of LesOdrRequest) // Request sends an ODR request to the LES network (implementation of LesOdrRequest)
func (r *ReceiptsRequest) Request(reqID uint64, peer *peer) error { func (r *ReceiptsRequest) Request(reqID uint64, peer *peer) error {
peer.Log().Debug("Requesting block receipts", "hash", r.Hash) peer.Log().Debug("Requesting block receipts", "hash", r.Hash)
return peer.RequestReceipts(reqID, r.GetCost(peer), []common.Hash{r.Hash}) return peer.RequestReceipts(reqID, []common.Hash{r.Hash})
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network
@ -208,7 +208,7 @@ func (r *TrieRequest) Request(reqID uint64, peer *peer) error {
AccKey: r.Id.AccKey, AccKey: r.Id.AccKey,
Key: r.Key, Key: r.Key,
} }
return peer.RequestProofs(reqID, r.GetCost(peer), []ProofReq{req}) return peer.RequestProofs(reqID, []ProofReq{req})
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network
@ -261,7 +261,7 @@ func (r *CodeRequest) Request(reqID uint64, peer *peer) error {
BHash: r.Id.BlockHash, BHash: r.Id.BlockHash,
AccKey: r.Id.AccKey, AccKey: r.Id.AccKey,
} }
return peer.RequestCode(reqID, r.GetCost(peer), []CodeReq{req}) return peer.RequestCode(reqID, []CodeReq{req})
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network
@ -343,7 +343,7 @@ func (r *ChtRequest) Request(reqID uint64, peer *peer) error {
Key: encNum[:], Key: encNum[:],
AuxReq: auxHeader, AuxReq: auxHeader,
} }
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req}) return peer.RequestHelperTrieProofs(reqID, []HelperTrieReq{req})
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network
@ -444,7 +444,7 @@ func (r *BloomRequest) Request(reqID uint64, peer *peer) error {
Key: common.CopyBytes(encNumber[:]), Key: common.CopyBytes(encNumber[:]),
} }
} }
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), reqs) return peer.RequestHelperTrieProofs(reqID, reqs)
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network
@ -501,7 +501,7 @@ func (r *TxStatusRequest) CanSend(peer *peer) bool {
// Request sends an ODR request to the LES network (implementation of LesOdrRequest) // Request sends an ODR request to the LES network (implementation of LesOdrRequest)
func (r *TxStatusRequest) Request(reqID uint64, peer *peer) error { func (r *TxStatusRequest) Request(reqID uint64, peer *peer) error {
peer.Log().Debug("Requesting transaction status", "count", len(r.Hashes)) peer.Log().Debug("Requesting transaction status", "count", len(r.Hashes))
return peer.RequestTxStatus(reqID, r.GetCost(peer), r.Hashes) return peer.RequestTxStatus(reqID, r.Hashes)
} }
// Valid processes an ODR request reply message from the LES network // Valid processes an ODR request reply message from the LES network

41
les/peer.go
View file

@ -86,7 +86,8 @@ type peer struct {
checkpoint params.TrustedCheckpoint checkpoint params.TrustedCheckpoint
checkpointNumber uint64 checkpointNumber uint64
id string id string
syncDrop mclock.Timer // Timed connection dropper if sync progress isn't validated in time
headInfo *announceData headInfo *announceData
lock sync.RWMutex lock sync.RWMutex
@ -296,7 +297,7 @@ func (p *peer) responseID() uint64 {
return p.responseCount return p.responseCount
} }
func sendRequest(w p2p.MsgWriter, msgcode, reqID, cost uint64, data interface{}) error { func sendRequest(w p2p.MsgWriter, msgcode, reqID uint64, data interface{}) error {
type req struct { type req struct {
ReqID uint64 ReqID uint64
Data interface{} Data interface{}
@ -445,60 +446,60 @@ func (p *peer) ReplyTxStatus(reqID uint64, stats []light.TxStatus) *reply {
// RequestHeadersByHash fetches a batch of blocks' headers corresponding to the // RequestHeadersByHash fetches a batch of blocks' headers corresponding to the
// specified header query, based on the hash of an origin block. // 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 { func (p *peer) RequestHeadersByHash(reqID 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) 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, GetBlockHeadersMsg, reqID, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
} }
// RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the // RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the
// specified header query, based on the number of an origin block. // 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 { func (p *peer) RequestHeadersByNumber(reqID, origin uint64, amount int, skip int, reverse bool) error {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse) 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, GetBlockHeadersMsg, reqID, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
} }
// RequestBodies fetches a batch of blocks' bodies corresponding to the hashes // RequestBodies fetches a batch of blocks' bodies corresponding to the hashes
// specified. // specified.
func (p *peer) RequestBodies(reqID, cost uint64, hashes []common.Hash) error { func (p *peer) RequestBodies(reqID uint64, hashes []common.Hash) error {
p.Log().Debug("Fetching batch of block bodies", "count", len(hashes)) p.Log().Debug("Fetching batch of block bodies", "count", len(hashes))
return sendRequest(p.rw, GetBlockBodiesMsg, reqID, cost, hashes) return sendRequest(p.rw, GetBlockBodiesMsg, reqID, hashes)
} }
// RequestCode fetches a batch of arbitrary data from a node's known state // RequestCode fetches a batch of arbitrary data from a node's known state
// data, corresponding to the specified hashes. // data, corresponding to the specified hashes.
func (p *peer) RequestCode(reqID, cost uint64, reqs []CodeReq) error { func (p *peer) RequestCode(reqID uint64, reqs []CodeReq) error {
p.Log().Debug("Fetching batch of codes", "count", len(reqs)) p.Log().Debug("Fetching batch of codes", "count", len(reqs))
return sendRequest(p.rw, GetCodeMsg, reqID, cost, reqs) return sendRequest(p.rw, GetCodeMsg, reqID, reqs)
} }
// RequestReceipts fetches a batch of transaction receipts from a remote node. // RequestReceipts fetches a batch of transaction receipts from a remote node.
func (p *peer) RequestReceipts(reqID, cost uint64, hashes []common.Hash) error { func (p *peer) RequestReceipts(reqID uint64, hashes []common.Hash) error {
p.Log().Debug("Fetching batch of receipts", "count", len(hashes)) p.Log().Debug("Fetching batch of receipts", "count", len(hashes))
return sendRequest(p.rw, GetReceiptsMsg, reqID, cost, hashes) return sendRequest(p.rw, GetReceiptsMsg, reqID, hashes)
} }
// RequestProofs fetches a batch of merkle proofs from a remote node. // 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 uint64, reqs []ProofReq) error {
p.Log().Debug("Fetching batch of proofs", "count", len(reqs)) p.Log().Debug("Fetching batch of proofs", "count", len(reqs))
return sendRequest(p.rw, GetProofsV2Msg, reqID, cost, reqs) return sendRequest(p.rw, GetProofsV2Msg, reqID, reqs)
} }
// RequestHelperTrieProofs fetches a batch of HelperTrie merkle proofs from a remote node. // 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 uint64, reqs []HelperTrieReq) error {
p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs)) p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs))
return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, cost, reqs) return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, reqs)
} }
// RequestTxStatus fetches a batch of transaction status records from a remote node. // RequestTxStatus fetches a batch of transaction status records from a remote node.
func (p *peer) RequestTxStatus(reqID, cost uint64, txHashes []common.Hash) error { func (p *peer) RequestTxStatus(reqID uint64, txHashes []common.Hash) error {
p.Log().Debug("Requesting transaction status", "count", len(txHashes)) p.Log().Debug("Requesting transaction status", "count", len(txHashes))
return sendRequest(p.rw, GetTxStatusMsg, reqID, cost, txHashes) return sendRequest(p.rw, GetTxStatusMsg, reqID, txHashes)
} }
// SendTxStatus creates a reply with a batch of transactions to be added to the remote transaction pool. // 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 { func (p *peer) SendTxs(reqID uint64, txs rlp.RawValue) error {
p.Log().Debug("Sending batch of transactions", "size", len(txs)) p.Log().Debug("Sending batch of transactions", "size", len(txs))
return sendRequest(p.rw, SendTxV2Msg, reqID, cost, txs) return sendRequest(p.rw, SendTxV2Msg, reqID, txs)
} }
type keyValueEntry struct { type keyValueEntry struct {

2
les/protocol.go
View file

@ -103,7 +103,6 @@ const (
ErrGenesisBlockMismatch ErrGenesisBlockMismatch
ErrNoStatusMsg ErrNoStatusMsg
ErrExtraStatusMsg ErrExtraStatusMsg
ErrSuspendedPeer
ErrUselessPeer ErrUselessPeer
ErrRequestRejected ErrRequestRejected
ErrUnexpectedResponse ErrUnexpectedResponse
@ -126,7 +125,6 @@ var errorToString = map[int]string{
ErrGenesisBlockMismatch: "Genesis block mismatch", ErrGenesisBlockMismatch: "Genesis block mismatch",
ErrNoStatusMsg: "No status message", ErrNoStatusMsg: "No status message",
ErrExtraStatusMsg: "Extra status message", ErrExtraStatusMsg: "Extra status message",
ErrSuspendedPeer: "Suspended peer",
ErrRequestRejected: "Request rejected", ErrRequestRejected: "Request rejected",
ErrUnexpectedResponse: "Unexpected response", ErrUnexpectedResponse: "Unexpected response",
ErrInvalidResponse: "Invalid response", ErrInvalidResponse: "Invalid response",

2
les/txrelay.go
View file

@ -136,7 +136,7 @@ func (self *lesTxRelay) send(txs types.Transactions, count int) {
peer := dp.(*peer) peer := dp.(*peer)
cost := peer.GetTxRelayCost(len(ll), len(enc)) cost := peer.GetTxRelayCost(len(ll), len(enc))
peer.fcServer.QueuedRequest(reqID, cost) peer.fcServer.QueuedRequest(reqID, cost)
return func() { peer.SendTxs(reqID, cost, enc) } return func() { peer.SendTxs(reqID, enc) }
}, },
} }
go self.retriever.retrieve(context.Background(), reqID, rq, func(p distPeer, msg *Msg) error { return nil }, self.stop) go self.retriever.retrieve(context.Background(), reqID, rq, func(p distPeer, msg *Msg) error { return nil }, self.stop)