woop/hmy/downloader/longrange_test.go

package downloader

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"testing"

	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
	"github.com/pkg/errors"
)

func TestDownloader_doLongRangeSync(t *testing.T) {
	targetBN := uint64(1000)
	bc := newTestBlockChain(1, nil)

	d := &Downloader{
		bc:           bc,
		ih:           &testInsertHelper{bc},
		syncProtocol: newTestSyncProtocol(targetBN, 32, nil),
		config: Config{
			Concurrency: 16,
			MinStreams:  16,
		},
		ctx: context.Background(),
	}
	synced, err := d.doLongRangeSync()
	if err != nil {
		t.Error(err)
	}
	if synced == 0 {
		t.Errorf("synced false")
	}
	if curNum := d.bc.CurrentBlock().NumberU64(); curNum != targetBN {
		t.Errorf("block number not expected: %v / %v", curNum, targetBN)
	}
}

func TestLrSyncIter_EstimateCurrentNumber(t *testing.T) {
	lsi := &lrSyncIter{
		p:   newTestSyncProtocol(100, 32, nil),
		ctx: context.Background(),
		config: Config{
			Concurrency: 16,
			MinStreams:  10,
		},
	}
	bn, err := lsi.estimateCurrentNumber()
	if err != nil {
		t.Error(err)
	}
	if bn != 100 {
		t.Errorf("unexpected block number: %v / %v", bn, 100)
	}
}

func TestGetBlocksManager_GetNextBatch(t *testing.T) {
	tests := []struct {
		gbm    *getBlocksManager
		expBNs []uint64
	}{
		{
			gbm: makeGetBlocksManager(
				10, 100, []uint64{9, 11, 12, 13},
				[]uint64{14, 15, 16}, []uint64{}, 0,
			),
			expBNs: []uint64{17, 18, 19, 20, 21, 22, 23, 24, 25, 26},
		},
		{
			gbm: makeGetBlocksManager(
				10, 100, []uint64{9, 13, 14, 15, 16},
				[]uint64{}, []uint64{10, 11, 12}, 0,
			),
			expBNs: []uint64{11, 12, 17, 18, 19, 20, 21, 22, 23, 24},
		},
		{
			gbm: makeGetBlocksManager(
				10, 100, []uint64{9, 13, 14, 15, 16},
				[]uint64{}, []uint64{10, 11, 12}, 120,
			),
			expBNs: []uint64{11, 12},
		},
		{
			gbm: makeGetBlocksManager(
				10, 100, []uint64{9, 13, 14, 15, 16},
				[]uint64{}, []uint64{}, 120,
			),
			expBNs: []uint64{},
		},
		{
			gbm: makeGetBlocksManager(
				10, 20, []uint64{9, 13, 14, 15, 16},
				[]uint64{}, []uint64{}, 0,
			),
			expBNs: []uint64{11, 12, 17, 18, 19, 20},
		},
		{
			gbm: makeGetBlocksManager(
				10, 100, []uint64{9, 13, 14, 15, 16},
				[]uint64{}, []uint64{}, 0,
			),
			expBNs: []uint64{11, 12, 17, 18, 19, 20, 21, 22, 23, 24},
		},
	}

	for i, test := range tests {
		if i < 4 {
			continue
		}
		batch := test.gbm.GetNextBatch()
		if len(test.expBNs) != len(batch) {
			t.Errorf("Test %v: unexpected size [%v] / [%v]", i, batch, test.expBNs)
		}
		for i := range test.expBNs {
			if test.expBNs[i] != batch[i] {
				t.Errorf("Test %v: [%v] / [%v]", i, batch, test.expBNs)
			}
		}
	}
}

func TestLrSyncIter_FetchAndInsertBlocks(t *testing.T) {
	targetBN := uint64(1000)
	chain := newTestBlockChain(0, nil)
	protocol := newTestSyncProtocol(targetBN, 32, nil)
	ctx, _ := context.WithCancel(context.Background())

	lsi := &lrSyncIter{
		bc:  chain,
		ih:  &testInsertHelper{chain},
		d:   &Downloader{},
		p:   protocol,
		gbm: nil,
		config: Config{
			Concurrency: 100,
		},
		ctx: ctx,
	}
	lsi.fetchAndInsertBlocks(targetBN)

	if err := fetchAndInsertBlocksResultCheck(lsi, targetBN, initStreamNum); err != nil {
		t.Error(err)
	}
}

// When FetchAndInsertBlocks, one request has an error
func TestLrSyncIter_FetchAndInsertBlocks_ErrRequest(t *testing.T) {
	targetBN := uint64(1000)
	var once sync.Once
	errHook := func(bn uint64) error {
		var err error
		once.Do(func() {
			err = errors.New("test error expected")
		})
		return err
	}
	chain := newTestBlockChain(0, nil)
	protocol := newTestSyncProtocol(targetBN, 32, errHook)
	ctx, _ := context.WithCancel(context.Background())

	lsi := &lrSyncIter{
		bc:  chain,
		ih:  &testInsertHelper{chain},
		d:   &Downloader{},
		p:   protocol,
		gbm: nil,
		config: Config{
			Concurrency: 100,
		},
		ctx: ctx,
	}
	lsi.fetchAndInsertBlocks(targetBN)

	if err := fetchAndInsertBlocksResultCheck(lsi, targetBN, initStreamNum-1); err != nil {
		t.Error(err)
	}
}

// When FetchAndInsertBlocks, one insertion has an error
func TestLrSyncIter_FetchAndInsertBlocks_ErrInsert(t *testing.T) {
	targetBN := uint64(1000)
	var once sync.Once
	errHook := func(bn uint64) error {
		var err error
		once.Do(func() {
			err = errors.New("test error expected")
		})
		return err
	}
	chain := newTestBlockChain(0, errHook)
	protocol := newTestSyncProtocol(targetBN, 32, nil)
	ctx, _ := context.WithCancel(context.Background())

	lsi := &lrSyncIter{
		bc:  chain,
		ih:  &testInsertHelper{chain},
		d:   &Downloader{},
		p:   protocol,
		gbm: nil,
		config: Config{
			Concurrency: 100,
		},
		ctx: ctx,
	}
	lsi.fetchAndInsertBlocks(targetBN)

	if err := fetchAndInsertBlocksResultCheck(lsi, targetBN, initStreamNum-1); err != nil {
		t.Error(err)
	}
}

// When FetchAndInsertBlocks, randomly error happens
func TestLrSyncIter_FetchAndInsertBlocks_RandomErr(t *testing.T) {
	targetBN := uint64(10000)
	rand.Seed(0)
	errHook := func(bn uint64) error {
		// 10% error happens
		if rand.Intn(10)%10 == 0 {
			return errors.New("error expected")
		}
		return nil
	}
	chain := newTestBlockChain(0, errHook)
	protocol := newTestSyncProtocol(targetBN, 32, errHook)
	ctx, _ := context.WithCancel(context.Background())

	lsi := &lrSyncIter{
		bc:  chain,
		ih:  &testInsertHelper{chain},
		d:   &Downloader{},
		p:   protocol,
		gbm: nil,
		config: Config{
			Concurrency: 100,
		},
		ctx: ctx,
	}
	lsi.fetchAndInsertBlocks(targetBN)

	if err := fetchAndInsertBlocksResultCheck(lsi, targetBN, minStreamNum); err != nil {
		t.Error(err)
	}
}

func fetchAndInsertBlocksResultCheck(lsi *lrSyncIter, targetBN uint64, expNumStreams int) error {
	if bn := lsi.bc.CurrentBlock().NumberU64(); bn != targetBN {
		return fmt.Errorf("did not reached targetBN: %v / %v", bn, targetBN)
	}
	lsi.gbm.lock.Lock()
	defer lsi.gbm.lock.Unlock()
	if len(lsi.gbm.processing) != 0 {
		return fmt.Errorf("not empty processing: %v", lsi.gbm.processing)
	}
	if len(lsi.gbm.requesting) != 0 {
		return fmt.Errorf("not empty requesting: %v", lsi.gbm.requesting)
	}
	if lsi.gbm.retries.length() != 0 {
		return fmt.Errorf("not empty retries: %v", lsi.gbm.retries)
	}
	if lsi.gbm.rq.length() != 0 {
		return fmt.Errorf("not empty result queue: %v", lsi.gbm.rq.results)
	}
	tsp := lsi.p.(*testSyncProtocol)
	if len(tsp.streamIDs) != expNumStreams {
		return fmt.Errorf("num streams not expected: %v / %v", len(tsp.streamIDs), expNumStreams)
	}
	return nil
}

func TestComputeBNMaxVote(t *testing.T) {
	tests := []struct {
		votes map[sttypes.StreamID]uint64
		exp   uint64
	}{
		{
			votes: map[sttypes.StreamID]uint64{
				makeStreamID(0): 10,
				makeStreamID(1): 10,
				makeStreamID(2): 20,
			},
			exp: 10,
		},
		{
			votes: map[sttypes.StreamID]uint64{
				makeStreamID(0): 10,
				makeStreamID(1): 20,
			},
			exp: 20,
		},
		{
			votes: map[sttypes.StreamID]uint64{
				makeStreamID(0): 20,
				makeStreamID(1): 10,
				makeStreamID(2): 20,
			},
			exp: 20,
		},
	}

	for i, test := range tests {
		res := computeBlockNumberByMaxVote(test.votes)
		if res != test.exp {
			t.Errorf("Test %v: unexpected bn %v / %v", i, res, test.exp)
		}
	}
}

func makeGetBlocksManager(curBN, targetBN uint64, requesting, processing, retries []uint64, sizeRQ int) *getBlocksManager {
	chain := newTestBlockChain(curBN, nil)
	requestingM := make(map[uint64]struct{})
	for _, bn := range requesting {
		requestingM[bn] = struct{}{}
	}
	processingM := make(map[uint64]struct{})
	for _, bn := range processing {
		processingM[bn] = struct{}{}
	}
	retriesPN := newPrioritizedNumbers()
	for _, retry := range retries {
		retriesPN.push(retry)
	}
	rq := newResultQueue()
	for i := uint64(0); i != uint64(sizeRQ); i++ {
		rq.addBlockResults(makeTestBlocks([]uint64{i + curBN}), "")
	}
	return &getBlocksManager{
		chain:      chain,
		targetBN:   targetBN,
		requesting: requestingM,
		processing: processingM,
		retries:    retriesPN,
		rq:         rq,
		resultC:    make(chan struct{}, 1),
	}
}