Merge branch 'master' of github.com:harmony-one/harmony into crosslink

hmy/api_backend.go

@@ -367,3 +367,8 @@ func (b *APIBackend) GetValidatorSelfDelegation(addr common.Address) *big.Int {
 func (b *APIBackend) GetShardState() (*shard.State, error) {
 	return b.hmy.BlockChain().ReadShardState(b.hmy.BlockChain().CurrentHeader().Epoch())
 }
+
+// GetCurrentStakingTransactionErrorSink ..
+func (b *APIBackend) GetCurrentStakingTransactionErrorSink() []staking.RPCTransactionError {
+	return b.hmy.nodeAPI.ErroredStakingTransactionSink()
+}

hmy/backend.go

@@ -52,11 +52,15 @@ type NodeAPI interface {
 	GetNonceOfAddress(address common.Address) uint64
 	GetTransactionsHistory(address, txType, order string) ([]common.Hash, error)
 	IsCurrentlyLeader() bool
+	ErroredStakingTransactionSink() []staking.RPCTransactionError
 }
 
 // New creates a new Harmony object (including the
 // initialisation of the common Harmony object)
-func New(nodeAPI NodeAPI, txPool *core.TxPool, cxPool *core.CxPool, eventMux *event.TypeMux, shardID uint32) (*Harmony, error) {
+func New(
+	nodeAPI NodeAPI, txPool *core.TxPool,
+	cxPool *core.CxPool, eventMux *event.TypeMux, shardID uint32,
+) (*Harmony, error) {
 	chainDb := nodeAPI.Blockchain().ChainDB()
 	hmy := &Harmony{
 		shutdownChan:   make(chan bool),

internal/chain/engine.go

@@ -11,6 +11,7 @@ import (
 	"github.com/harmony-one/bls/ffi/go/bls"
 	"github.com/harmony-one/harmony/block"
 	"github.com/harmony-one/harmony/consensus/engine"
+	"github.com/harmony-one/harmony/consensus/quorum"
 	"github.com/harmony-one/harmony/consensus/reward"
 	"github.com/harmony-one/harmony/core/state"
 	"github.com/harmony-one/harmony/core/types"
@@ -206,18 +207,43 @@ func (e *engineImpl) VerifySeal(chain engine.ChainReader, header *block.Header)
 	payload := append(sig[:], header.LastCommitBitmap()...)
 	aggSig, mask, err := ReadSignatureBitmapByPublicKeys(payload, publicKeys)
 	if err != nil {
-		return ctxerror.New("[VerifySeal] Unable to deserialize the LastCommitSignature and LastCommitBitmap in Block Header").WithCause(err)
+		return ctxerror.New(
+			"[VerifySeal] Unable to deserialize the LastCommitSignature" +
+				" and LastCommitBitmap in Block Header",
+		).WithCause(err)
 	}
 	parentHash := header.ParentHash()
 	parentHeader := chain.GetHeader(parentHash, header.Number().Uint64()-1)
+	if chain.Config().IsStaking(parentHeader.Epoch()) {
+		slotList, err := chain.ReadShardState(parentHeader.Epoch())
+		if err != nil {
+			return errors.Wrapf(err, "cannot decoded shard state")
+		}
+		d := quorum.NewDecider(quorum.SuperMajorityStake)
+		d.SetShardIDProvider(func() (uint32, error) {
+			return parentHeader.ShardID(), nil
+		})
+		d.SetMyPublicKeyProvider(func() (*bls.PublicKey, error) {
+			return nil, nil
+		})
+		d.SetVoters(slotList.FindCommitteeByID(parentHeader.ShardID()).Slots)
+		if !d.IsQuorumAchievedByMask(mask) {
+			return ctxerror.New(
+				"[VerifySeal] Not enough voting power in LastCommitSignature from Block Header",
+			)
+		}
+	} else {
 		parentQuorum, err := QuorumForBlock(chain, parentHeader, false)
 		if err != nil {
 			return errors.Wrapf(err,
 				"cannot calculate quorum for block %s", header.Number())
 		}
 		if count := utils.CountOneBits(mask.Bitmap); count < int64(parentQuorum) {
-		return ctxerror.New("[VerifySeal] Not enough signature in LastCommitSignature from Block Header",
-			"need", parentQuorum, "got", count)
+			return ctxerror.New(
+				"[VerifySeal] Not enough signature in LastCommitSignature from Block Header",
+				"need", parentQuorum, "got", count,
+			)
+		}
 	}
 
 	blockNumHash := make([]byte, 8)
@@ -259,7 +285,7 @@ func (e *engineImpl) Finalize(
 			wrapper := state.GetStakingInfo(validator)
 			if wrapper != nil {
 				for i := range wrapper.Delegations {
-					delegation := wrapper.Delegations[i]
+					delegation := &wrapper.Delegations[i]
 					totalWithdraw := delegation.RemoveUnlockedUndelegations(header.Epoch())
 					state.AddBalance(delegation.DelegatorAddress, totalWithdraw)
 				}
@@ -318,18 +344,37 @@ func (e *engineImpl) VerifyHeaderWithSignature(chain engine.ChainReader, header
 	if err != nil {
 		return ctxerror.New("[VerifyHeaderWithSignature] Unable to deserialize the commitSignature and commitBitmap in Block Header").WithCause(err)
 	}
-
 	hash := header.Hash()
-	quorum, err := QuorumForBlock(chain, header, reCalculate)
+
+	if e := header.Epoch(); chain.Config().IsStaking(e) {
+		slotList, err := chain.ReadShardState(e)
+		if err != nil {
+			return errors.Wrapf(err, "cannot read shard state")
+		}
+		d := quorum.NewDecider(quorum.SuperMajorityStake)
+		d.SetShardIDProvider(func() (uint32, error) {
+			return header.ShardID(), nil
+		})
+		d.SetMyPublicKeyProvider(func() (*bls.PublicKey, error) {
+			return nil, nil
+		})
+		d.SetVoters(slotList.FindCommitteeByID(header.ShardID()).Slots)
+		if !d.IsQuorumAchievedByMask(mask) {
+			return ctxerror.New(
+				"[VerifySeal] Not enough voting power in commitSignature from Block Header",
+			)
+		}
+	} else {
+		quorumCount, err := QuorumForBlock(chain, header, reCalculate)
 		if err != nil {
 			return errors.Wrapf(err,
 				"cannot calculate quorum for block %s", header.Number())
 		}
-	if count := utils.CountOneBits(mask.Bitmap); count < int64(quorum) {
+		if count := utils.CountOneBits(mask.Bitmap); count < int64(quorumCount) {
 			return ctxerror.New("[VerifyHeaderWithSignature] Not enough signature in commitSignature from Block Header",
-			"need", quorum, "got", count)
+				"need", quorumCount, "got", count)
+		}
 	}
-
 	blockNumHash := make([]byte, 8)
 	binary.LittleEndian.PutUint64(blockNumHash, header.Number().Uint64())
 	commitPayload := append(blockNumHash, hash[:]...)

internal/hmyapi/backend.go

@@ -81,6 +81,7 @@ type Backend interface {
 	GetDelegationsByDelegator(delegator common.Address) ([]common.Address, []*staking.Delegation)
 	GetValidatorSelfDelegation(addr common.Address) *big.Int
 	GetShardState() (*shard.State, error)
+	GetCurrentStakingTransactionErrorSink() []staking.RPCTransactionError
 }
 
 // GetAPIs returns all the APIs.

internal/hmyapi/harmony.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/harmony-one/harmony/api/proto"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
+	staking "github.com/harmony-one/harmony/staking/types"
 )
 
 // PublicHarmonyAPI provides an API to access Harmony related information.
@@ -65,3 +66,8 @@ func (s *PublicHarmonyAPI) GetNodeMetadata() NodeMetadata {
 		s.b.GetShardID(),
 	}
 }
+
+// GetCurrentTransactionErrorSink ..
+func (s *PublicHarmonyAPI) GetCurrentTransactionErrorSink() []staking.RPCTransactionError {
+	return s.b.GetCurrentStakingTransactionErrorSink()
+}

node/node.go

@@ -216,6 +216,11 @@ type Node struct {
 
 	// last time consensus reached for metrics
 	lastConsensusTime int64
+	// Last 1024 staking transaction error, only in memory
+	errorSink struct {
+		sync.Mutex
+		failedTxns []staking.RPCTransactionError
+	}
 }
 
 // Blockchain returns the blockchain for the node's current shard.
@@ -390,6 +395,10 @@ func (node *Node) GetSyncID() [SyncIDLength]byte {
 func New(host p2p.Host, consensusObj *consensus.Consensus,
 	chainDBFactory shardchain.DBFactory, isArchival bool) *Node {
 	node := Node{}
+	node.errorSink = struct {
+		sync.Mutex
+		failedTxns []staking.RPCTransactionError
+	}{}
 
 	node.syncFreq = SyncFrequency
 	node.beaconSyncFreq = SyncFrequency

node/node_cross_link.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	maxPendingCrossLinkSize = 300
+	maxPendingCrossLinkSize = 1000
 	crossLinkBatchSize      = 10
 )
 
@@ -160,6 +160,12 @@ func (node *Node) VerifyCrossLink(cl types.CrossLink) error {
 	}
 
 	decider := quorum.NewDecider(quorum.SuperMajorityStake)
+	decider.SetShardIDProvider(func() (uint32, error) {
+		return cl.ShardID(), nil
+	})
+	decider.SetMyPublicKeyProvider(func() (*bls.PublicKey, error) {
+		return nil, nil
+	})
 	if _, err := decider.SetVoters(committee.Slots); err != nil {
 		return ctxerror.New("[VerifyCrossLink] Cannot SetVoters for committee", "shardID", cl.ShardID())
 	}

node/node_explorer.go

@@ -48,11 +48,8 @@ func (node *Node) ExplorerMessageHandler(payload []byte) {
 			return
 		}
 
-		// check has 2f+1 signatures
-		need := node.Consensus.Decider.TwoThirdsSignersCount()
-		if count := utils.CountOneBits(mask.Bitmap); count < need {
-			utils.Logger().Error().Int64("need", need).Int64("have", count).
-				Msg("[Explorer] not have enough signature")
+		if node.Consensus.Decider.IsQuorumAchievedByMask(mask) {
+			utils.Logger().Error().Msg("[Explorer] not have enough signature power")
 			return
 		}
 

node/node_handler_test.go

@@ -3,18 +3,17 @@ package node
 import (
 	"testing"
 
-	"github.com/harmony-one/harmony/core/types"
-	types2 "github.com/harmony-one/harmony/staking/types"
-
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/harmony-one/harmony/consensus"
 	"github.com/harmony-one/harmony/consensus/quorum"
+	"github.com/harmony-one/harmony/core/types"
 	"github.com/harmony-one/harmony/crypto/bls"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/harmony-one/harmony/p2p"
 	"github.com/harmony-one/harmony/p2p/p2pimpl"
 	"github.com/harmony-one/harmony/shard"
+	staking "github.com/harmony-one/harmony/staking/types"
 )
 
 func TestAddNewBlock(t *testing.T) {
@@ -37,8 +36,8 @@ func TestAddNewBlock(t *testing.T) {
 	node := New(host, consensus, testDBFactory, false)
 
 	txs := make(map[common.Address]types.Transactions)
-	stks := types2.StakingTransactions{}
-	node.Worker.CommitTransactions(txs, stks, common.Address{})
+	stks := staking.StakingTransactions{}
+	node.Worker.CommitTransactions(txs, stks, common.Address{}, func(staking.RPCTransactionError) {})
 	block, _ := node.Worker.FinalizeNewBlock([]byte{}, []byte{}, 0, common.Address{}, nil, nil)
 
 	_, err = node.Blockchain().InsertChain([]*types.Block{block}, true)
@@ -70,8 +69,8 @@ func TestVerifyNewBlock(t *testing.T) {
 	node := New(host, consensus, testDBFactory, false)
 
 	txs := make(map[common.Address]types.Transactions)
-	stks := types2.StakingTransactions{}
-	node.Worker.CommitTransactions(txs, stks, common.Address{})
+	stks := staking.StakingTransactions{}
+	node.Worker.CommitTransactions(txs, stks, common.Address{}, func(staking.RPCTransactionError) {})
 	block, _ := node.Worker.FinalizeNewBlock([]byte{}, []byte{}, 0, common.Address{}, nil, nil)
 
 	if err := node.VerifyNewBlock(block); err != nil {

node/node_newblock.go

@@ -5,15 +5,12 @@ import (
 	"strings"
 	"time"
 
-	"github.com/harmony-one/harmony/shard"
-
-	"github.com/harmony-one/harmony/core/rawdb"
-
-	types2 "github.com/harmony-one/harmony/staking/types"
-
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/harmony-one/harmony/core/rawdb"
 	"github.com/harmony-one/harmony/core/types"
 	"github.com/harmony-one/harmony/internal/utils"
+	"github.com/harmony-one/harmony/shard"
+	staking "github.com/harmony-one/harmony/staking/types"
 )
 
 // Constants of proposing a new block
@@ -113,19 +110,31 @@ func (node *Node) proposeNewBlock() (*types.Block, error) {
 	}
 
 	// TODO: integrate staking transaction into tx pool
-	pendingStakingTransactions := types2.StakingTransactions{}
+	pendingStakingTransactions := staking.StakingTransactions{}
 	// Only process staking transactions after pre-staking epoch happened.
 	if node.Blockchain().Config().IsPreStaking(node.Worker.GetCurrentHeader().Epoch()) {
 		node.pendingStakingTxMutex.Lock()
 		for _, tx := range node.pendingStakingTransactions {
 			pendingStakingTransactions = append(pendingStakingTransactions, tx)
 		}
-		node.pendingStakingTransactions = make(map[common.Hash]*types2.StakingTransaction)
+		node.pendingStakingTransactions = make(map[common.Hash]*staking.StakingTransaction)
 		node.pendingStakingTxMutex.Unlock()
 	}
 
-	if err := node.Worker.CommitTransactions(pending, pendingStakingTransactions, beneficiary); err != nil {
-		utils.Logger().Error().Err(err).Msg("[proposeNewBlock] cannot commit transactions")
+	if err := node.Worker.CommitTransactions(
+		pending, pendingStakingTransactions, beneficiary,
+		func(payload staking.RPCTransactionError) {
+			const maxSize = 1024
+			node.errorSink.Lock()
+			if l := len(node.errorSink.failedTxns); l >= maxSize {
+				node.errorSink.failedTxns = append(node.errorSink.failedTxns[1:], payload)
+			} else {
+				node.errorSink.failedTxns = append(node.errorSink.failedTxns, payload)
+			}
+			node.errorSink.Unlock()
+		},
+	); err != nil {
+		utils.Logger().Error().Err(err).Msg("cannot commit transactions")
 		return nil, err
 	}
 

node/rpc.go

@@ -8,12 +8,12 @@ import (
 
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/rpc"
-
 	"github.com/harmony-one/harmony/hmy"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 	"github.com/harmony-one/harmony/internal/hmyapi"
 	"github.com/harmony-one/harmony/internal/hmyapi/filters"
 	"github.com/harmony-one/harmony/internal/utils"
+	staking "github.com/harmony-one/harmony/staking/types"
 )
 
 const (
@@ -49,10 +49,19 @@ func (node *Node) IsCurrentlyLeader() bool {
 	return node.Consensus.IsLeader()
 }
 
+// ErroredStakingTransactionSink is the inmemory failed staking transactions this node has
+func (node *Node) ErroredStakingTransactionSink() []staking.RPCTransactionError {
+	node.errorSink.Lock()
+	defer node.errorSink.Unlock()
+	return node.errorSink.failedTxns
+}
+
 // StartRPC start RPC service
 func (node *Node) StartRPC(nodePort string) error {
 	// Gather all the possible APIs to surface
-	harmony, _ = hmy.New(node, node.TxPool, node.CxPool, new(event.TypeMux), node.Consensus.ShardID)
+	harmony, _ = hmy.New(
+		node, node.TxPool, node.CxPool, new(event.TypeMux), node.Consensus.ShardID,
+	)
 
 	apis := node.APIs()
 

node/worker/worker.go

@@ -26,10 +26,8 @@ import (
 // environment is the worker's current environment and holds all of the current state information.
 type environment struct {
 	signer     types.Signer
-
 	state      *state.DB     // apply state changes here
 	gasPool    *core.GasPool // available gas used to pack transactions
-
 	header     *block.Header
 	txs        []*types.Transaction
 	stakingTxs staking.StakingTransactions
@@ -45,15 +43,17 @@ type Worker struct {
 	factory  blockfactory.Factory
 	chain    *core.BlockChain
 	current  *environment // An environment for current running cycle.
-
 	engine   consensus_engine.Engine
-
 	gasFloor uint64
 	gasCeil  uint64
 }
 
 // CommitTransactions commits transactions for new block.
-func (w *Worker) CommitTransactions(pendingNormal map[common.Address]types.Transactions, pendingStaking staking.StakingTransactions, coinbase common.Address) error {
+func (w *Worker) CommitTransactions(
+	pendingNormal map[common.Address]types.Transactions,
+	pendingStaking staking.StakingTransactions, coinbase common.Address,
+	stkingTxErrorSink func(staking.RPCTransactionError),
+) error {
 
 	if w.current.gasPool == nil {
 		w.current.gasPool = new(core.GasPool).AddGas(w.current.header.GasLimit())
@@ -131,11 +131,21 @@ func (w *Worker) CommitTransactions(pendingNormal map[common.Address]types.Trans
 	if w.chain.ShardID() == shard.BeaconChainShardID {
 		for _, tx := range pendingStaking {
 			logs, err := w.commitStakingTransaction(tx, coinbase)
-			if err != nil {
-				utils.Logger().Error().Err(err).Str("stakingTxId", tx.Hash().Hex()).Msg("Commit staking transaction error")
+			if txID := tx.Hash().Hex(); err != nil {
+				stkingTxErrorSink(staking.RPCTransactionError{
+					TxHashID:             txID,
+					StakingDirective:     tx.StakingType().String(),
+					TimestampOfRejection: time.Now().Unix(),
+					ErrMessage:           err.Error(),
+				})
+				utils.Logger().Error().Err(err).
+					Str("stakingTxId", txID).
+					Msg("Commit staking transaction error")
 			} else {
 				coalescedLogs = append(coalescedLogs, logs...)
-				utils.Logger().Info().Str("stakingTxId", tx.Hash().Hex()).Uint64("txGasLimit", tx.Gas()).Msg("StakingTransaction gas limit info")
+				utils.Logger().Info().Str("stakingTxId", tx.Hash().Hex()).
+					Uint64("txGasLimit", tx.Gas()).
+					Msg("StakingTransaction gas limit info")
 			}
 		}
 	}
@@ -145,11 +155,15 @@ func (w *Worker) CommitTransactions(pendingNormal map[common.Address]types.Trans
 	return nil
 }
 
-func (w *Worker) commitStakingTransaction(tx *staking.StakingTransaction, coinbase common.Address) ([]*types.Log, error) {
+func (w *Worker) commitStakingTransaction(
+	tx *staking.StakingTransaction, coinbase common.Address,
+) ([]*types.Log, error) {
 	snap := w.current.state.Snapshot()
 	gasUsed := w.current.header.GasUsed()
-	receipt, _, err :=
-		core.ApplyStakingTransaction(w.config, w.chain, &coinbase, w.current.gasPool, w.current.state, w.current.header, tx, &gasUsed, vm.Config{})
+	receipt, _, err := core.ApplyStakingTransaction(
+		w.config, w.chain, &coinbase, w.current.gasPool,
+		w.current.state, w.current.header, tx, &gasUsed, vm.Config{},
+	)
 	w.current.header.SetGasUsed(gasUsed)
 	if err != nil {
 		w.current.state.RevertToSnapshot(snap)

node/worker/worker_test.go

@@ -6,7 +6,6 @@ import (
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
-
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
 	blockfactory "github.com/harmony-one/harmony/block/factory"
@@ -16,6 +15,7 @@ import (
 	"github.com/harmony-one/harmony/core/vm"
 	chain2 "github.com/harmony-one/harmony/internal/chain"
 	"github.com/harmony-one/harmony/internal/params"
+	staking "github.com/harmony-one/harmony/staking/types"
 )
 
 var (
@@ -78,7 +78,7 @@ func TestCommitTransactions(t *testing.T) {
 	// Commit the tx to the worker
 	txs := make(map[common.Address]types.Transactions)
 	txs[testBankAddress] = types.Transactions{tx}
-	err := worker.CommitTransactions(txs, nil, testBankAddress)
+	err := worker.CommitTransactions(txs, nil, testBankAddress, func(staking.RPCTransactionError) {})
 	if err != nil {
 		t.Error(err)
 	}

staking/types/delegation.go

@@ -25,7 +25,7 @@ type Delegation struct {
 	DelegatorAddress common.Address `json:"delegator_address" yaml:"delegator_address"`
 	Amount           *big.Int       `json:"amount" yaml:"amount"`
 	Reward           *big.Int       `json:"reward" yaml:"reward"`
-	Undelegations    []*Undelegation `json:"undelegations" yaml:"undelegations"`
+	Undelegations    []Undelegation `json:"undelegations" yaml:"undelegations"`
 }
 
 // Undelegation represents one undelegation entry
@@ -70,7 +70,7 @@ func (d *Delegation) Undelegate(epoch *big.Int, amt *big.Int) error {
 
 	if !exist {
 		item := Undelegation{amt, epoch}
-		d.Undelegations = append(d.Undelegations, &item)
+		d.Undelegations = append(d.Undelegations, item)
 
 		// Always sort the undelegate by epoch in increasing order
 		sort.SliceStable(
@@ -94,7 +94,7 @@ func (d *Delegation) TotalInUndelegation() *big.Int {
 // DeleteEntry - delete an entry from the undelegation
 // Opimize it
 func (d *Delegation) DeleteEntry(epoch *big.Int) {
-	entries := []*Undelegation{}
+	entries := []Undelegation{}
 	for i, entry := range d.Undelegations {
 		if entry.Epoch.Cmp(epoch) == 0 {
 			entries = append(d.Undelegations[:i], d.Undelegations[i+1:]...)

staking/types/rpc-result.go

@@ -0,0 +1,9 @@
+package types
+
+// RPCTransactionError ..
+type RPCTransactionError struct {
+	TxHashID             string `json:"tx-hash-id"`
+	StakingDirective     string `json:"directive-kind"`
+	TimestampOfRejection int64  `json:"time-at-rejection"`
+	ErrMessage           string `json:"error-message"`
+}

test/chain/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/harmony-one/harmony/crypto/hash"
 	"github.com/harmony-one/harmony/internal/params"
 	pkgworker "github.com/harmony-one/harmony/node/worker"
+	staking "github.com/harmony-one/harmony/staking/types"
 )
 
 const (
@@ -129,7 +130,7 @@ func fundFaucetContract(chain *core.BlockChain) {
 	txmap := make(map[common.Address]types.Transactions)
 	txmap[FaucetAddress] = txs
 
-	err := contractworker.CommitTransactions(txmap, nil, testUserAddress)
+	err := contractworker.CommitTransactions(txmap, nil, testUserAddress, func(staking.RPCTransactionError) {})
 	if err != nil {
 		fmt.Println(err)
 	}
@@ -170,7 +171,7 @@ func callFaucetContractToFundAnAddress(chain *core.BlockChain) {
 	txmap := make(map[common.Address]types.Transactions)
 	txmap[FaucetAddress] = types.Transactions{callfaucettx}
 
-	err = contractworker.CommitTransactions(txmap, nil, testUserAddress)
+	err = contractworker.CommitTransactions(txmap, nil, testUserAddress, func(staking.RPCTransactionError) {})
 	if err != nil {
 		fmt.Println(err)
 	}