statecmp.go

package main

import (
	"context"
	"encoding/json"
	"flag"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/pkg/errors"

	"github.com/wavesplatform/gowaves/pkg/client"
	"github.com/wavesplatform/gowaves/pkg/crypto"
	"github.com/wavesplatform/gowaves/pkg/logging"
	"github.com/wavesplatform/gowaves/pkg/proto"
)

const (
	defaultScheme = "http"
)

func checkAndUpdateURL(s string) (string, error) {
	var u *url.URL
	var err error
	if strings.Contains(s, "//") {
		u, err = url.Parse(s)
	} else {
		u, err = url.Parse("//" + s)
	}
	if err != nil {
		return "", errors.Wrapf(err, "failed to parse URL %s", s)
	}
	if u.Scheme == "" {
		u.Scheme = defaultScheme
	}
	if u.Scheme != "http" && u.Scheme != "https" {
		return "", errors.Errorf("unsupported URL scheme %s", u.Scheme)
	}
	return u.String(), nil
}

func loadStateHash(ctx context.Context, cl *client.Client, height uint64, tries int) (proto.StateHash, error) {
	ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
	defer cancel()
	var sh proto.StateHash
	var err error
	for range tries {
		sh, _, err = cl.Debug.StateHash(ctx, height)
		if err == nil {
			return sh, nil
		}
	}
	return nil, err
}

type printer struct {
	lock sync.Mutex
}

func (p *printer) printDifferentResults(height uint64, res map[crypto.Digest]*nodesGroup) {
	p.lock.Lock()
	defer p.lock.Unlock()

	for sh, nodes := range res {
		hashJs, err := json.Marshal(sh)
		if err != nil {
			panic(err)
		}
		slog.Info("Following state hashes are different",
			"height", height, "stateHash", string(hashJs), "nodes", nodes.nodes)
	}
}

type stateHashInfo struct {
	sh   proto.StateHash
	node string
}

type hashResult struct {
	res stateHashInfo
	err error
}

type nodesGroup struct {
	nodes []string
	sh    proto.StateHash
}

func newNodesGroup(first string, sh proto.StateHash) *nodesGroup {
	nodes := make([]string, 1)
	nodes[0] = first
	return &nodesGroup{nodes: nodes, sh: sh}
}

func (ng *nodesGroup) addNode(node string) {
	ng.nodes = append(ng.nodes, node)
}

func manageHeight(
	ctx context.Context, height uint64, nodes []string, clients []*client.Client, p *printer, tries int,
) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	results := make(chan hashResult, len(clients))
	for i, cl := range clients {
		node := nodes[i]
		go func(cl *client.Client, node string) {
			sh, err := loadStateHash(ctx, cl, height, tries)
			res := hashResult{
				res: stateHashInfo{
					sh:   sh,
					node: node,
				},
				err: err,
			}
			results <- res
		}(cl, node)
	}
	differentResults := make(map[crypto.Digest]*nodesGroup)
	for range clients {
		hr := <-results
		if hr.err != nil {
			cancel()
			return hr.err
		}
		sh := hr.res.sh
		ng, ok := differentResults[sh.GetSumHash()]
		if !ok {
			differentResults[sh.GetSumHash()] = newNodesGroup(hr.res.node, sh)
		} else {
			ng.addNode(hr.res.node)
		}
	}
	if len(differentResults) != 1 {
		p.printDifferentResults(height, differentResults)
	}
	return nil
}

func download(
	heightChan chan uint64, errChan chan error, nodes []string, clients []*client.Client, goroutines, tries int,
) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	p := &printer{}
	var wg sync.WaitGroup
	for range goroutines {
		wg.Go(func() {
			for height := range heightChan {
				if err := manageHeight(ctx, height, nodes, clients, p, tries); err != nil {
					cancel()
					slog.Error("Failed to load hashes", logging.Error(err))
					errChan <- err
					break
				}
			}
		})
	}
	wg.Wait()
}

func main() {
	const (
		defaultStartHeight      = 1
		defaultEndHeight        = 2
		defaultGoroutinesNumber = 15
		defaultTriesNumber      = 5
		minNodesNumber          = 2
	)
	var (
		lp            = logging.Parameters{}
		nodesStr      = flag.String("nodes", "", "Addresses of nodes; comma separated.")
		startHeight   = flag.Int("start-height", defaultStartHeight, "Start height.")
		endHeight     = flag.Int("end-height", defaultEndHeight, "End height.")
		goroutinesNum = flag.Int("goroutines-num", defaultGoroutinesNumber,
			"Number of goroutines that will run for downloading state hashes.")
		tries = flag.Int("tries-num", defaultTriesNumber, "Number of tries to download.")
	)
	lp.Initialize()
	flag.Parse()

	if err := lp.Parse(); err != nil {
		slog.Error("Failed to parse application parameters", logging.Error(err))
		os.Exit(1)
	}

	slog.SetDefault(slog.New(logging.DefaultHandler(lp)))
	if *endHeight <= *startHeight {
		slog.Error("End height must be greater than start height",
			"endHeight", *endHeight, "startHeight", *startHeight)
		os.Exit(1)
	}

	nodes := strings.FieldsFunc(*nodesStr, func(r rune) bool { return r == ',' })
	if len(nodes) < minNodesNumber {
		slog.Error("At least 2 nodes are required for state comparison", "nodes", *nodesStr)
		os.Exit(1)
	}
	clients := make([]*client.Client, len(nodes))
	for i, nu := range nodes {
		u, err := checkAndUpdateURL(nu)
		if err != nil {
			slog.Error("Failed to update URL", logging.Error(err))
			os.Exit(1)
		}
		clients[i], err = client.NewClient(client.Options{BaseUrl: u, Client: &http.Client{}})
		if err != nil {
			slog.Error("Failed to create client", slog.String("URL", u), logging.Error(err))
			os.Exit(1)
		}
	}

	heightChan := make(chan uint64)
	errChan := make(chan error, *goroutinesNum)
	var wg sync.WaitGroup
	wg.Go(func() {
		download(heightChan, errChan, nodes, clients, *goroutinesNum, *tries)
	})
	for h := uint64(*startHeight); h < uint64(*endHeight); h++ {
		gotErr := false
		select {
		case heightChan <- h:
		case <-errChan:
			gotErr = true
		}
		if gotErr {
			break
		}
	}
	close(heightChan)
	wg.Wait()
	slog.Info("Finished to compare states.")
}