all: make indexer configurable (#17188)

1 files changed, 92 insertions, 87 deletions

diff --git a/les/handler_test.go b/les/handler_test.go
index 31aad3ed4..43be7f41b 100644
--- a/les/handler_test.go
+++ b/les/handler_test.go
@@ -51,10 +51,9 @@ func TestGetBlockHeadersLes1(t *testing.T) { testGetBlockHeaders(t, 1) }
 func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) }
 
 func testGetBlockHeaders(t *testing.T, protocol int) {
-	pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, nil, nil, nil, ethdb.NewMemDatabase())
-	bc := pm.blockchain.(*core.BlockChain)
-	peer, _ := newTestPeer(t, "peer", protocol, pm, true)
-	defer peer.close()
+	server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil)
+	defer tearDown()
+	bc := server.pm.blockchain.(*core.BlockChain)
 
 	// Create a "random" unknown hash for testing
 	var unknown common.Hash
@@ -167,9 +166,9 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
 		}
 		// Send the hash request and verify the response
 		reqID++
-		cost := peer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount))
-		sendRequest(peer.app, GetBlockHeadersMsg, reqID, cost, tt.query)
-		if err := expectResponse(peer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil {
+		cost := server.tPeer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount))
+		sendRequest(server.tPeer.app, GetBlockHeadersMsg, reqID, cost, tt.query)
+		if err := expectResponse(server.tPeer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil {
 			t.Errorf("test %d: headers mismatch: %v", i, err)
 		}
 	}
@@ -180,10 +179,9 @@ func TestGetBlockBodiesLes1(t *testing.T) { testGetBlockBodies(t, 1) }
 func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) }
 
 func testGetBlockBodies(t *testing.T, protocol int) {
-	pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, nil, nil, nil, ethdb.NewMemDatabase())
-	bc := pm.blockchain.(*core.BlockChain)
-	peer, _ := newTestPeer(t, "peer", protocol, pm, true)
-	defer peer.close()
+	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
 	limit := MaxBodyFetch
@@ -243,9 +241,9 @@ func testGetBlockBodies(t *testing.T, protocol int) {
 		}
 		reqID++
 		// Send the hash request and verify the response
-		cost := peer.GetRequestCost(GetBlockBodiesMsg, len(hashes))
-		sendRequest(peer.app, GetBlockBodiesMsg, reqID, cost, hashes)
-		if err := expectResponse(peer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil {
+		cost := server.tPeer.GetRequestCost(GetBlockBodiesMsg, len(hashes))
+		sendRequest(server.tPeer.app, GetBlockBodiesMsg, reqID, cost, hashes)
+		if err := expectResponse(server.tPeer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil {
 			t.Errorf("test %d: bodies mismatch: %v", i, err)
 		}
 	}
@@ -257,10 +255,9 @@ func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) }
 
 func testGetCode(t *testing.T, protocol int) {
 	// Assemble the test environment
-	pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, ethdb.NewMemDatabase())
-	bc := pm.blockchain.(*core.BlockChain)
-	peer, _ := newTestPeer(t, "peer", protocol, pm, true)
-	defer peer.close()
+	server, tearDown := newServerEnv(t, 4, protocol, nil)
+	defer tearDown()
+	bc := server.pm.blockchain.(*core.BlockChain)
 
 	var codereqs []*CodeReq
 	var codes [][]byte
@@ -277,9 +274,9 @@ func testGetCode(t *testing.T, protocol int) {
 		}
 	}
 
-	cost := peer.GetRequestCost(GetCodeMsg, len(codereqs))
-	sendRequest(peer.app, GetCodeMsg, 42, cost, codereqs)
-	if err := expectResponse(peer.app, CodeMsg, 42, testBufLimit, codes); err != nil {
+	cost := server.tPeer.GetRequestCost(GetCodeMsg, len(codereqs))
+	sendRequest(server.tPeer.app, GetCodeMsg, 42, cost, codereqs)
+	if err := expectResponse(server.tPeer.app, CodeMsg, 42, testBufLimit, codes); err != nil {
 		t.Errorf("codes mismatch: %v", err)
 	}
 }
@@ -290,11 +287,9 @@ func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) }
 
 func testGetReceipt(t *testing.T, protocol int) {
 	// Assemble the test environment
-	db := ethdb.NewMemDatabase()
-	pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
-	bc := pm.blockchain.(*core.BlockChain)
-	peer, _ := newTestPeer(t, "peer", protocol, pm, true)
-	defer peer.close()
+	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
 	hashes, receipts := []common.Hash{}, []types.Receipts{}
@@ -302,12 +297,12 @@ func testGetReceipt(t *testing.T, protocol int) {
 		block := bc.GetBlockByNumber(i)
 
 		hashes = append(hashes, block.Hash())
-		receipts = append(receipts, rawdb.ReadReceipts(db, block.Hash(), block.NumberU64()))
+		receipts = append(receipts, rawdb.ReadReceipts(server.db, block.Hash(), block.NumberU64()))
 	}
 	// Send the hash request and verify the response
-	cost := peer.GetRequestCost(GetReceiptsMsg, len(hashes))
-	sendRequest(peer.app, GetReceiptsMsg, 42, cost, hashes)
-	if err := expectResponse(peer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil {
+	cost := server.tPeer.GetRequestCost(GetReceiptsMsg, len(hashes))
+	sendRequest(server.tPeer.app, GetReceiptsMsg, 42, cost, hashes)
+	if err := expectResponse(server.tPeer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil {
 		t.Errorf("receipts mismatch: %v", err)
 	}
 }
@@ -318,11 +313,9 @@ func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) }
 
 func testGetProofs(t *testing.T, protocol int) {
 	// Assemble the test environment
-	db := ethdb.NewMemDatabase()
-	pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
-	bc := pm.blockchain.(*core.BlockChain)
-	peer, _ := newTestPeer(t, "peer", protocol, pm, true)
-	defer peer.close()
+	server, tearDown := newServerEnv(t, 4, protocol, nil)
+	defer tearDown()
+	bc := server.pm.blockchain.(*core.BlockChain)
 
 	var (
 		proofreqs []ProofReq
@@ -334,7 +327,7 @@ func testGetProofs(t *testing.T, protocol int) {
 	for i := uint64(0); i <= bc.CurrentBlock().NumberU64(); i++ {
 		header := bc.GetHeaderByNumber(i)
 		root := header.Root
-		trie, _ := trie.New(root, trie.NewDatabase(db))
+		trie, _ := trie.New(root, trie.NewDatabase(server.db))
 
 		for _, acc := range accounts {
 			req := ProofReq{
@@ -356,15 +349,15 @@ func testGetProofs(t *testing.T, protocol int) {
 	// Send the proof request and verify the response
 	switch protocol {
 	case 1:
-		cost := peer.GetRequestCost(GetProofsV1Msg, len(proofreqs))
-		sendRequest(peer.app, GetProofsV1Msg, 42, cost, proofreqs)
-		if err := expectResponse(peer.app, ProofsV1Msg, 42, testBufLimit, proofsV1); err != nil {
+		cost := server.tPeer.GetRequestCost(GetProofsV1Msg, len(proofreqs))
+		sendRequest(server.tPeer.app, GetProofsV1Msg, 42, cost, proofreqs)
+		if err := expectResponse(server.tPeer.app, ProofsV1Msg, 42, testBufLimit, proofsV1); err != nil {
 			t.Errorf("proofs mismatch: %v", err)
 		}
 	case 2:
-		cost := peer.GetRequestCost(GetProofsV2Msg, len(proofreqs))
-		sendRequest(peer.app, GetProofsV2Msg, 42, cost, proofreqs)
-		if err := expectResponse(peer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil {
+		cost := server.tPeer.GetRequestCost(GetProofsV2Msg, len(proofreqs))
+		sendRequest(server.tPeer.app, GetProofsV2Msg, 42, cost, proofreqs)
+		if err := expectResponse(server.tPeer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil {
 			t.Errorf("proofs mismatch: %v", err)
 		}
 	}
@@ -375,28 +368,33 @@ func TestGetCHTProofsLes1(t *testing.T) { testGetCHTProofs(t, 1) }
 func TestGetCHTProofsLes2(t *testing.T) { testGetCHTProofs(t, 2) }
 
 func testGetCHTProofs(t *testing.T, protocol int) {
-	// Figure out the client's CHT frequency
-	frequency := uint64(light.CHTFrequencyClient)
-	if protocol == 1 {
-		frequency = uint64(light.CHTFrequencyServer)
+	config := light.TestServerIndexerConfig
+	frequency := config.ChtSize
+	if protocol == 2 {
+		frequency = config.PairChtSize
 	}
-	// Assemble the test environment
-	db := ethdb.NewMemDatabase()
-	pm := newTestProtocolManagerMust(t, false, int(frequency)+light.HelperTrieProcessConfirmations, testChainGen, nil, nil, db)
-	bc := pm.blockchain.(*core.BlockChain)
-	peer, _ := newTestPeer(t, "peer", protocol, pm, true)
-	defer peer.close()
 
-	// Wait a while for the CHT indexer to process the new headers
-	time.Sleep(100 * time.Millisecond * time.Duration(frequency/light.CHTFrequencyServer)) // Chain indexer throttling
-	time.Sleep(250 * time.Millisecond)                                                     // CI tester slack
+	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.ChtConfirms), protocol, waitIndexers)
+	defer tearDown()
+	bc := server.pm.blockchain.(*core.BlockChain)
 
 	// Assemble the proofs from the different protocols
-	header := bc.GetHeaderByNumber(frequency)
+	header := bc.GetHeaderByNumber(frequency - 1)
 	rlp, _ := rlp.EncodeToBytes(header)
 
 	key := make([]byte, 8)
-	binary.BigEndian.PutUint64(key, frequency)
+	binary.BigEndian.PutUint64(key, frequency-1)
 
 	proofsV1 := []ChtResp{{
 		Header: header,
@@ -406,41 +404,41 @@ func testGetCHTProofs(t *testing.T, protocol int) {
 	}
 	switch protocol {
 	case 1:
-		root := light.GetChtRoot(db, 0, bc.GetHeaderByNumber(frequency-1).Hash())
-		trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.ChtTablePrefix)))
+		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
 		trie.Prove(key, 0, &proof)
 		proofsV1[0].Proof = proof
 
 	case 2:
-		root := light.GetChtV2Root(db, 0, bc.GetHeaderByNumber(frequency-1).Hash())
-		trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.ChtTablePrefix)))
+		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: frequency,
+		ChtNum:   frequency / config.ChtSize,
+		BlockNum: frequency - 1,
 	}}
 	requestsV2 := []HelperTrieReq{{
 		Type:    htCanonical,
-		TrieIdx: 0,
+		TrieIdx: frequency/config.PairChtSize - 1,
 		Key:     key,
 		AuxReq:  auxHeader,
 	}}
 	// Send the proof request and verify the response
 	switch protocol {
 	case 1:
-		cost := peer.GetRequestCost(GetHeaderProofsMsg, len(requestsV1))
-		sendRequest(peer.app, GetHeaderProofsMsg, 42, cost, requestsV1)
-		if err := expectResponse(peer.app, HeaderProofsMsg, 42, testBufLimit, proofsV1); err != nil {
+		cost := server.tPeer.GetRequestCost(GetHeaderProofsMsg, len(requestsV1))
+		sendRequest(server.tPeer.app, GetHeaderProofsMsg, 42, cost, requestsV1)
+		if err := expectResponse(server.tPeer.app, HeaderProofsMsg, 42, testBufLimit, proofsV1); err != nil {
 			t.Errorf("proofs mismatch: %v", err)
 		}
 	case 2:
-		cost := peer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2))
-		sendRequest(peer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2)
-		if err := expectResponse(peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil {
+		cost := server.tPeer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2))
+		sendRequest(server.tPeer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2)
+		if err := expectResponse(server.tPeer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil {
 			t.Errorf("proofs mismatch: %v", err)
 		}
 	}
@@ -448,24 +446,31 @@ func testGetCHTProofs(t *testing.T, protocol int) {
 
 // Tests that bloombits proofs can be correctly retrieved.
 func TestGetBloombitsProofs(t *testing.T) {
-	// Assemble the test environment
-	db := ethdb.NewMemDatabase()
-	pm := newTestProtocolManagerMust(t, false, light.BloomTrieFrequency+256, testChainGen, nil, nil, db)
-	bc := pm.blockchain.(*core.BlockChain)
-	peer, _ := newTestPeer(t, "peer", 2, pm, true)
-	defer peer.close()
-
-	// Wait a while for the bloombits indexer to process the new headers
-	time.Sleep(100 * time.Millisecond * time.Duration(light.BloomTrieFrequency/4096)) // Chain indexer throttling
-	time.Sleep(250 * time.Millisecond)                                                // CI tester slack
+	config := light.TestServerIndexerConfig
+
+	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.BloomTrieConfirms), 2, waitIndexers)
+	defer tearDown()
+	bc := server.pm.blockchain.(*core.BlockChain)
 
 	// Request and verify each bit of the bloom bits proofs
 	for bit := 0; bit < 2048; bit++ {
-		// Assemble therequest and proofs for the bloombits
+		// Assemble the request and proofs for the bloombits
 		key := make([]byte, 10)
 
 		binary.BigEndian.PutUint16(key[:2], uint16(bit))
-		binary.BigEndian.PutUint64(key[2:], uint64(light.BloomTrieFrequency))
+		// Only the first bloom section has data.
+		binary.BigEndian.PutUint64(key[2:], 0)
 
 		requests := []HelperTrieReq{{
 			Type:    htBloomBits,
@@ -474,14 +479,14 @@ func TestGetBloombitsProofs(t *testing.T) {
 		}}
 		var proofs HelperTrieResps
 
-		root := light.GetBloomTrieRoot(db, 0, bc.GetHeaderByNumber(light.BloomTrieFrequency-1).Hash())
-		trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.BloomTrieTablePrefix)))
+		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)
 
 		// Send the proof request and verify the response
-		cost := peer.GetRequestCost(GetHelperTrieProofsMsg, len(requests))
-		sendRequest(peer.app, GetHelperTrieProofsMsg, 42, cost, requests)
-		if err := expectResponse(peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil {
+		cost := server.tPeer.GetRequestCost(GetHelperTrieProofsMsg, len(requests))
+		sendRequest(server.tPeer.app, GetHelperTrieProofsMsg, 42, cost, requests)
+		if err := expectResponse(server.tPeer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil {
 			t.Errorf("bit %d: proofs mismatch: %v", bit, err)
 		}
 	}