diff --git a/les/handler.go b/les/handler.go index dabeb83cf6..0faf5c04e9 100644 --- a/les/handler.go +++ b/les/handler.go @@ -1149,9 +1149,9 @@ func (pm *ProtocolManager) getAccount(statedb *state.StateDB, root, hash common. func (pm *ProtocolManager) getHelperTrie(id uint, idx uint64) (common.Hash, string) { switch id { case htCanonical: - idxV2 := (idx+1)*(pm.indexerConfig.PairChtSize/pm.indexerConfig.ChtSize) - 1 - sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.indexerConfig.PairChtSize-1) - return light.GetChtRoot(pm.chainDb, idxV2, sectionHead), light.ChtTablePrefix + idxV1 := (idx+1)*(pm.indexerConfig.PairChtSize/pm.indexerConfig.ChtSize) - 1 + sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idxV1+1)*pm.indexerConfig.ChtSize-1) + return light.GetChtRoot(pm.chainDb, idxV1, sectionHead), light.ChtTablePrefix case htBloomBits: sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*pm.indexerConfig.BloomTrieSize-1) return light.GetBloomTrieRoot(pm.chainDb, idx, sectionHead), light.BloomTrieTablePrefix diff --git a/les/handler_test.go b/les/handler_test.go index 81297d1a8f..09d96d57aa 100644 --- a/les/handler_test.go +++ b/les/handler_test.go @@ -51,12 +51,8 @@ func TestGetBlockHeadersLes1(t *testing.T) { testGetBlockHeaders(t, 1) } func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) } func testGetBlockHeaders(t *testing.T, protocol int) { - server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, 0) - defer func() { - if tearDown != nil { - tearDown() - } - }() + server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil) + defer tearDown() bc := server.pm.blockchain.(*core.BlockChain) // Create a "random" unknown hash for testing @@ -183,12 +179,8 @@ func TestGetBlockBodiesLes1(t *testing.T) { testGetBlockBodies(t, 1) } func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) } func testGetBlockBodies(t *testing.T, protocol int) { - server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, 0) - defer func() { - if tearDown != nil { - tearDown() - } - }() + server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, nil) + defer tearDown() bc := server.pm.blockchain.(*core.BlockChain) // Create a batch of tests for various scenarios @@ -263,12 +255,8 @@ func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) } func testGetCode(t *testing.T, protocol int) { // Assemble the test environment - server, tearDown := newServerEnv(t, 4, protocol, 0) - defer func() { - if tearDown != nil { - tearDown() - } - }() + server, tearDown := newServerEnv(t, 4, protocol, nil) + defer tearDown() bc := server.pm.blockchain.(*core.BlockChain) var codereqs []*CodeReq @@ -299,12 +287,8 @@ func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) } func testGetReceipt(t *testing.T, protocol int) { // Assemble the test environment - server, tearDown := newServerEnv(t, 4, protocol, 0) - defer func() { - if tearDown != nil { - tearDown() - } - }() + server, tearDown := newServerEnv(t, 4, protocol, nil) + defer tearDown() bc := server.pm.blockchain.(*core.BlockChain) // Collect the hashes to request, and the response to expect @@ -329,12 +313,8 @@ func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) } func testGetProofs(t *testing.T, protocol int) { // Assemble the test environment - server, tearDown := newServerEnv(t, 4, protocol, 0) - defer func() { - if tearDown != nil { - tearDown() - } - }() + server, tearDown := newServerEnv(t, 4, protocol, nil) + defer tearDown() bc := server.pm.blockchain.(*core.BlockChain) var ( @@ -391,23 +371,30 @@ func testGetCHTProofs(t *testing.T, protocol int) { config := light.TestServerIndexerConfig frequency := config.ChtSize if protocol == 2 { - frequency = config.ChtSize * 8 + frequency = config.PairChtSize } - server, tearDown := newServerEnv(t, int(frequency+config.ChtConfirm), protocol, 1) - defer func() { - if tearDown != nil { - tearDown() + waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) { + expectSections := frequency / config.ChtSize + for { + cs, _, _ := cIndexer.Sections() + bs, _, _ := bIndexer.Sections() + if cs >= expectSections && bs >= expectSections { + break + } + time.Sleep(10 * time.Millisecond) } - }() + } + server, tearDown := newServerEnv(t, int(frequency+config.ChtConfirm), protocol, waitIndexers) + defer tearDown() bc := server.pm.blockchain.(*core.BlockChain) // Assemble the proofs from the different protocols - header := bc.GetHeaderByNumber(config.ChtSize - 1) + header := bc.GetHeaderByNumber(frequency - 1) rlp, _ := rlp.EncodeToBytes(header) key := make([]byte, 8) - binary.BigEndian.PutUint64(key, config.ChtSize-1) + binary.BigEndian.PutUint64(key, frequency-1) proofsV1 := []ChtResp{{ Header: header, @@ -417,7 +404,7 @@ func testGetCHTProofs(t *testing.T, protocol int) { } switch protocol { case 1: - root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(config.ChtSize-1).Hash()) + root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(frequency-1).Hash()) trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) var proof light.NodeList @@ -425,18 +412,18 @@ func testGetCHTProofs(t *testing.T, protocol int) { proofsV1[0].Proof = proof case 2: - root := light.GetChtRoot(server.db, (config.ChtSize/config.PairChtSize)-1, bc.GetHeaderByNumber(config.ChtSize-1).Hash()) + root := light.GetChtRoot(server.db, (frequency/config.ChtSize)-1, bc.GetHeaderByNumber(frequency-1).Hash()) trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.ChtTablePrefix))) trie.Prove(key, 0, &proofsV2.Proofs) } // Assemble the requests for the different protocols requestsV1 := []ChtReq{{ - ChtNum: 1, - BlockNum: config.ChtSize - 1, + ChtNum: frequency / config.ChtSize, + BlockNum: frequency - 1, }} requestsV2 := []HelperTrieReq{{ Type: htCanonical, - TrieIdx: 0, + TrieIdx: frequency/config.PairChtSize - 1, Key: key, AuxReq: auxHeader, }} @@ -460,12 +447,20 @@ func testGetCHTProofs(t *testing.T, protocol int) { // Tests that bloombits proofs can be correctly retrieved. func TestGetBloombitsProofs(t *testing.T) { config := light.TestServerIndexerConfig - server, tearDown := newServerEnv(t, int(config.BloomSize+config.BloomConfirm), 2, 1) - defer func() { - if tearDown != nil { - tearDown() + + waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) { + for { + cs, _, _ := cIndexer.Sections() + bs, _, _ := bIndexer.Sections() + bts, _, _ := btIndexer.Sections() + if cs >= 8 && bs >= 8 && bts >= 1 { + break + } + time.Sleep(10 * time.Millisecond) } - }() + } + server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirm), 2, waitIndexers) + defer tearDown() bc := server.pm.blockchain.(*core.BlockChain) // Request and verify each bit of the bloom bits proofs @@ -474,7 +469,8 @@ func TestGetBloombitsProofs(t *testing.T) { key := make([]byte, 10) binary.BigEndian.PutUint16(key[:2], uint16(bit)) - binary.BigEndian.PutUint64(key[2:], config.BloomSize-1) + // Only the first bloom section has data. + binary.BigEndian.PutUint64(key[2:], 0) requests := []HelperTrieReq{{ Type: htBloomBits, @@ -483,7 +479,7 @@ func TestGetBloombitsProofs(t *testing.T) { }} var proofs HelperTrieResps - root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomSize-1).Hash()) + root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomTrieSize-1).Hash()) trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(server.db, light.BloomTrieTablePrefix))) trie.Prove(key, 0, &proofs.Proofs) diff --git a/les/helper_test.go b/les/helper_test.go index d17efa62ed..f193a988f9 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -348,7 +348,7 @@ type TestEntity struct { } // newServerEnv creates a server testing environment with a connected test peer for testing purpose. -func newServerEnv(t *testing.T, blocks int, protocol int, processSections uint64) (*TestEntity, func()) { +func newServerEnv(t *testing.T, blocks int, protocol int, waitIndexers func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer)) (*TestEntity, func()) { db := ethdb.NewMemDatabase() cIndexer, bIndexer, btIndexer := testIndexers(db, light.TestServerIndexerConfig) @@ -359,16 +359,10 @@ func newServerEnv(t *testing.T, blocks int, protocol int, processSections uint64 bIndexer.Start(pm.blockchain.(*core.BlockChain)) // Wait until indexers generate enough index data. - if processSections > 0 { - for { - cs, _, _ := cIndexer.Sections() - bs, _, _ := bIndexer.Sections() - if cs >= processSections && bs >= processSections { - break - } - time.Sleep(10 * time.Millisecond) - } + if waitIndexers != nil { + waitIndexers(cIndexer, bIndexer, btIndexer) } + return &TestEntity{ db: db, tPeer: peer, @@ -386,7 +380,7 @@ func newServerEnv(t *testing.T, blocks int, protocol int, processSections uint64 // newClientServerEnv creates a client/server arch environment with a connected les server and light client pair // for testing purpose. -func newClientServerEnv(t *testing.T, blocks int, protocol int) (*TestEntity, *TestEntity, func()) { +func newClientServerEnv(t *testing.T, blocks int, protocol int, waitIndexers func(*core.ChainIndexer, *core.ChainIndexer, *core.ChainIndexer), newPeer bool) (*TestEntity, *TestEntity, func()) { db, ldb := ethdb.NewMemDatabase(), ethdb.NewMemDatabase() peers, lPeers := newPeerSet(), newPeerSet() @@ -411,13 +405,24 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int) (*TestEntity, *T startIndexers(false, pm) startIndexers(true, lpm) - peer, err1, lPeer, err2 := newTestPeerPair("peer", protocol, pm, lpm) - 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) + // Execute wait until function if it is specified. + if waitIndexers != nil { + waitIndexers(cIndexer, bIndexer, btIndexer) + } + + var ( + peer, lPeer *peer + err1, err2 <-chan error + ) + if newPeer { + peer, err1, lPeer, err2 = newTestPeerPair("peer", protocol, pm, lpm) + 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) + } } return &TestEntity{ diff --git a/les/odr_requests.go b/les/odr_requests.go index f34cac0ae3..96bd5a488a 100644 --- a/les/odr_requests.go +++ b/les/odr_requests.go @@ -387,10 +387,10 @@ func (r *ChtRequest) Request(reqID uint64, peer *peer, config *light.IndexerConf } blockNum := binary.BigEndian.Uint64(req.Key) // convert HelperTrie request to old CHT request - reqsV1 = ChtReq{ChtNum: (req.TrieIdx+1)*(config.ChtSize/config.PairChtSize) - 1, BlockNum: blockNum, FromLevel: req.FromLevel} - return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []interface{}{reqsV1}) + reqsV1 = ChtReq{ChtNum: (req.TrieIdx + 1) * (config.ChtSize / config.PairChtSize), BlockNum: blockNum, FromLevel: req.FromLevel} + return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []ChtReq{reqsV1}) case lpv2: - return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []interface{}{req}) + return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req}) default: panic(nil) } @@ -517,7 +517,7 @@ func (r *BloomRequest) Request(reqID uint64, peer *peer, config *light.IndexerCo Key: common.CopyBytes(encNumber[:]), } } - return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []interface{}{reqs}) + return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), reqs) } // Valid processes an ODR request reply message from the LES network diff --git a/les/odr_test.go b/les/odr_test.go index f041927139..e6458adf56 100644 --- a/les/odr_test.go +++ b/les/odr_test.go @@ -162,12 +162,8 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai // testOdr tests odr requests whose validation guaranteed by block headers. func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) { // Assemble the test environment - server, client, tearDown := newClientServerEnv(t, 4, protocol) - defer func() { - if tearDown != nil { - tearDown() - } - }() + server, client, tearDown := newClientServerEnv(t, 4, protocol, nil, true) + defer tearDown() client.pm.synchronise(client.rPeer) test := func(expFail uint64) { diff --git a/les/peer.go b/les/peer.go index 87c213cd5e..f4d6c53c75 100644 --- a/les/peer.go +++ b/les/peer.go @@ -35,9 +35,10 @@ import ( ) var ( - errClosed = errors.New("peer set is closed") - errAlreadyRegistered = errors.New("peer is already registered") - errNotRegistered = errors.New("peer is not registered") + errClosed = errors.New("peer set is closed") + errAlreadyRegistered = errors.New("peer is already registered") + errNotRegistered = errors.New("peer is not registered") + errInvalidHelpTrieReq = errors.New("invalid help trie request") ) const maxResponseErrors = 50 // number of invalid responses tolerated (makes the protocol less brittle but still avoids spam) @@ -283,12 +284,20 @@ func (p *peer) RequestProofs(reqID, cost uint64, reqs []ProofReq) error { } // RequestHelperTrieProofs fetches a batch of HelperTrie merkle proofs from a remote node. -func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []interface{}) error { - p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs)) +func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, data interface{}) error { switch p.version { case lpv1: + reqs, ok := data.([]ChtReq) + if !ok { + return errInvalidHelpTrieReq + } + p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs)) return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqs) case lpv2: + reqs, ok := data.([]HelperTrieReq) + if !ok { + return errInvalidHelpTrieReq + } return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, cost, reqs) default: panic(nil) diff --git a/les/request_test.go b/les/request_test.go index e465187a87..f02c2a3d76 100644 --- a/les/request_test.go +++ b/les/request_test.go @@ -83,12 +83,8 @@ func tfCodeAccess(db ethdb.Database, bhash common.Hash, num uint64) light.OdrReq func testAccess(t *testing.T, protocol int, fn accessTestFn) { // Assemble the test environment - server, client, tearDown := newClientServerEnv(t, 4, protocol) - defer func() { - if tearDown != nil { - tearDown() - } - }() + server, client, tearDown := newClientServerEnv(t, 4, protocol, nil, true) + defer tearDown() client.pm.synchronise(client.rPeer) test := func(expFail uint64) { diff --git a/light/postprocess.go b/light/postprocess.go index 7c71911057..f708c7868b 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -335,6 +335,5 @@ func (b *BloomTrieIndexerBackend) Commit() error { sectionHead := b.sectionHeads[b.bloomTrieRatio-1] log.Info("Storing bloom trie", "section", b.section, "head", sectionHead, "root", root, "compression", float64(compSize)/float64(decompSize)) StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root) - return nil }