neo-go/pkg/services/stateroot/validators.go

package stateroot

import (
	"time"

	"github.com/nspcc-dev/neo-go/pkg/core/state"
	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
	"github.com/nspcc-dev/neo-go/pkg/io"
	"github.com/nspcc-dev/neo-go/pkg/network/payload"
	"github.com/nspcc-dev/neo-go/pkg/vm/emit"
	"github.com/nspcc-dev/neo-go/pkg/wallet"
	"go.uber.org/zap"
)

const (
	voteValidEndInc      = 10
	firstVoteResendDelay = 3 * time.Second
)

// Name returns service name.
func (s *service) Name() string {
	return "stateroot"
}

// Start runs service instance in a separate goroutine.
// The service only starts once, subsequent calls to Start are no-op.
func (s *service) Start() {
	if !s.started.CompareAndSwap(false, true) {
		return
	}
	s.log.Info("starting state validation service")
	go s.run()
}

func (s *service) run() {
	s.chain.SubscribeForBlocks(s.blockCh)
runloop:
	for {
		select {
		case b := <-s.blockCh:
			r, err := s.GetStateRoot(b.Index)
			if err != nil {
				s.log.Error("can't get state root for new block", zap.Error(err))
			} else if err := s.signAndSend(r); err != nil {
				s.log.Error("can't sign or send state root", zap.Error(err))
			}
			s.srMtx.Lock()
			delete(s.incompleteRoots, b.Index-voteValidEndInc)
			s.srMtx.Unlock()
		case <-s.stopCh:
			break runloop
		}
	}
	s.chain.UnsubscribeFromBlocks(s.blockCh)
drainloop:
	for {
		select {
		case <-s.blockCh:
		default:
			break drainloop
		}
	}
	close(s.blockCh)
	close(s.done)
}

// Shutdown stops the service. It can only be called once, subsequent calls
// to Shutdown on the same instance are no-op. The instance that was stopped can
// not be started again by calling Start (use a new instance if needed).
func (s *service) Shutdown() {
	if !s.started.CompareAndSwap(true, false) {
		return
	}
	s.log.Info("stopping state validation service")
	close(s.stopCh)
	<-s.done
	if s.wallet != nil {
		s.wallet.Close()
	}
	_ = s.log.Sync()
}

func (s *service) signAndSend(r *state.MPTRoot) error {
	if !s.MainCfg.Enabled {
		return nil
	}

	myIndex, acc := s.getAccount()
	if acc == nil {
		return nil
	}

	sig := acc.SignHashable(s.Network, r)
	incRoot := s.getIncompleteRoot(r.Index, myIndex)
	incRoot.Lock()
	defer incRoot.Unlock()
	incRoot.root = r
	incRoot.addSignature(acc.PublicKey(), sig)
	incRoot.reverify(s.Network)
	s.trySendRoot(incRoot, acc)

	msg := NewMessage(VoteT, &Vote{
		ValidatorIndex: int32(myIndex),
		Height:         r.Index,
		Signature:      sig,
	})

	w := io.NewBufBinWriter()
	msg.EncodeBinary(w.BinWriter)
	if w.Err != nil {
		return w.Err
	}
	e := &payload.Extensible{
		Category:        Category,
		ValidBlockStart: r.Index,
		ValidBlockEnd:   r.Index + voteValidEndInc,
		Sender:          acc.ScriptHash(),
		Data:            w.Bytes(),
		Witness: transaction.Witness{
			VerificationScript: acc.GetVerificationScript(),
		},
	}
	sig = acc.SignHashable(s.Network, e)
	buf := io.NewBufBinWriter()
	emit.Bytes(buf.BinWriter, sig)
	e.Witness.InvocationScript = buf.Bytes()
	incRoot.myVote = e
	incRoot.retries = -1
	s.sendVote(incRoot)
	return nil
}

// sendVote attempts to send a vote if it's still valid and if stateroot message
// has not been sent yet. It must be called with the ir locked.
func (s *service) sendVote(ir *incompleteRoot) {
	if ir.isSent || ir.retries >= s.maxRetries ||
		s.chain.HeaderHeight() >= ir.myVote.ValidBlockEnd {
		return
	}
	s.relayExtensible(ir.myVote)
	delay := firstVoteResendDelay
	if ir.retries > 0 {
		delay = s.timePerBlock << ir.retries
	}
	_ = time.AfterFunc(delay, func() {
		ir.Lock()
		s.sendVote(ir)
		ir.Unlock()
	})
	ir.retries++
}

// getAccount returns the current index and account for the node running this service.
func (s *service) getAccount() (byte, *wallet.Account) {
	s.accMtx.RLock()
	defer s.accMtx.RUnlock()
	return s.myIndex, s.acc
}