Merge pull request #95 from dappledger/develop

dev->master

chain/app/evm/evm.go

@@ -393,47 +393,55 @@ func (app *EVMApp) OnCommit(height, round int64, block *gtypes.Block) (interface
 	}, nil
 }
 
-func (app *EVMApp) CheckTx(bs []byte) error {
+func (app *EVMApp) GetAddressFromTx(tx *etypes.Transaction) (from common.Address, err error) {
+	from, err = etypes.Sender(app.Signer, tx)
+	return
+}
+
+func (app *EVMApp) CheckTx(bs []byte) (from common.Address,nonce uint64, err error) {
 	tx := &etypes.Transaction{}
-	err := rlp.DecodeBytes(bs, tx)
+	err = rlp.DecodeBytes(bs, &tx)
 	if err != nil {
-		return err
+		return
 	}
-	from, _ := etypes.Sender(app.Signer, tx)
-
+	from, _ = etypes.Sender(app.Signer, tx)
 	app.stateMtx.Lock()
 	defer app.stateMtx.Unlock()
 	// Last but not least check for nonce errors
-	nonce := tx.Nonce()
+	nonce = tx.Nonce()
 	getNonce := app.state.GetNonce(from)
 	if getNonce > nonce {
 		txhash := gtypes.Tx(bs).Hash()
-		return fmt.Errorf("nonce(%d) different with getNonce(%d), transaction already exists %v", nonce, getNonce, hex.EncodeToString(txhash))
+		err = fmt.Errorf("nonce(%d) different with getNonce(%d), transaction already exists %v", nonce, getNonce, hex.EncodeToString(txhash))
+		return
 	}
 
 	// Transactor should have enough funds to cover the costs
 	// cost == V + GP * GL
 	if app.state.GetBalance(from).Cmp(tx.Cost()) < 0 {
-		return fmt.Errorf("not enough funds")
+		err = fmt.Errorf("not enough funds")
+		return
 	}
 
 	txType := common.Bytes2Hex(tx.Data())
 
 	if strings.HasPrefix(txType, common.Bytes2Hex(rtypes.KVTxType)) {
 		txData := tx.Data()[len(rtypes.KVTxType):]
 		kvData := &rtypes.KV{}
-		if err := rlp.DecodeBytes(txData, kvData); err != nil {
-			return fmt.Errorf("rlp decode to kv error %s", err.Error())
+		if err = rlp.DecodeBytes(txData, kvData); err != nil {
+			err = fmt.Errorf("rlp decode to kv error %s", err.Error())
+			return
 		}
 		if len(kvData.Key) > MaxKey || len(kvData.Value) > MaxValue {
-			return fmt.Errorf("key or value too big,MaxKey:%v,MaxValue:%v", MaxKey, MaxValue)
+			err = fmt.Errorf("key or value too big,MaxKey:%v,MaxValue:%v", MaxKey, MaxValue)
+			return
 		}
 		if ok, _ := app.stateDb.Has(append(KvPrefix, kvData.Key...)); ok {
-			return fmt.Errorf("duplicate key :%v", kvData.Key)
+			err = fmt.Errorf("duplicate key :%v", kvData.Key)
+			return
 		}
 	}
-
-	return nil
+	return
 }
 
 func (app *EVMApp) SaveReceipts() ([]byte, error) {
@@ -515,6 +523,8 @@ func (app *EVMApp) Query(query []byte) (res gtypes.Result) {
 		res = app.queryKey(load)
 	case rtypes.QueryType_Key_Prefix:
 		res = app.queryKeyWithPrefix(load)
+	case rtypes.QueryType_Pending_Nonce:
+		res = app.queryPendingNonce(load)
 	default:
 		res = gtypes.NewError(gtypes.CodeType_BaseInvalidInput, "unimplemented query")
 	}
@@ -606,6 +616,21 @@ func makeETHHeader(header *gtypes.Header) *etypes.Header {
 	}
 }
 
+func (app *EVMApp) queryPendingNonce(addrBytes []byte) gtypes.Result {
+	if len(addrBytes) != 20 {
+		return gtypes.NewError(gtypes.CodeType_BaseInvalidInput, "Invalid address")
+	}
+	nonce, err := app.GetTxPool().GetPendingMaxNonce(addrBytes)
+	if err != nil {
+		return gtypes.NewError(gtypes.CodeType_UnknownRequest, err.Error())
+	}
+	data, err := rlp.EncodeToBytes(nonce)
+	if err != nil {
+		log.Warn("query error", zap.Error(err))
+	}
+	return gtypes.NewResultOK(data, "")
+}
+
 func (app *EVMApp) queryNonce(addrBytes []byte) gtypes.Result {
 	if len(addrBytes) != 20 {
 		return gtypes.NewError(gtypes.CodeType_BaseInvalidInput, "Invalid address")

chain/app/evm/tx_pool.go

@@ -177,6 +177,29 @@ func (tp *ethTxPool) ReceiveTx(rawTx types.Tx) error {
 	return nil
 }
 
+func (tp *ethTxPool) GetPendingMaxNonce(bytAddress []byte) (uint64, error) {
+	tp.Lock()
+	defer tp.Unlock()
+	address := common.BytesToAddress(bytAddress)
+	txWSMap, ok := tp.waiting[address]
+	if ok && txWSMap != nil {
+		pendingMaxNonce := tp.unSafeGetPendingMaxNonce(address)
+		if pendingMaxNonce != txWSMap.MinNonce() {
+			return pendingMaxNonce, nil
+		}
+		return txWSMap.MaxNonce() + 1, nil
+	}
+	return tp.unSafeGetPendingMaxNonce(address), nil
+}
+
+func (tp *ethTxPool) unSafeGetPendingMaxNonce(address common.Address) uint64 {
+	txSMap, ok := tp.pending[address]
+	if ok && txSMap != nil {
+		return txSMap.MaxNonce() + 1
+	}
+	return tp.safeGetNonce(address)
+}
+
 // receive and handle adminOp txs
 func (tp *ethTxPool) handleAdminOP(tx types.Tx) error {
 	tp.Lock()

chain/app/evm/tx_sort.go

@@ -128,6 +128,19 @@ func (m *txSortedMap) MaxNonce() uint64 {
 	return maxNonce
 }
 
+// return min nonce in txSortedMap, call from empty m will cause a panic.
+func (m *txSortedMap) MinNonce() uint64 {
+	var sortedTxs types.Transactions
+	if m.cache != nil {
+		sortedTxs = m.cache
+	} else {
+		sortedTxs = m.Flatten()
+	}
+
+	minNonce := (sortedTxs)[0].Nonce()
+	return minNonce
+}
+
 // Remove deletes a transaction from the maintained map, returning whether the
 // transaction was found.
 func (m *txSortedMap) Remove(nonce uint64) bool {

chain/core/routes.go

@@ -210,28 +210,36 @@ func (h *rpcHandler) UnsafeFlushMempool() (*gtypes.ResultUnsafeFlushMempool, err
 	return &gtypes.ResultUnsafeFlushMempool{}, nil
 }
 
-func (h *rpcHandler) BroadcastTx(tx []byte) (*gtypes.ResultBroadcastTx, error) {
-	if err := h.node.Application.CheckTx(tx); err != nil {
-		return nil, err
+func (h *rpcHandler) BroadcastTx(tx []byte) (result *gtypes.ResultBroadcastTx, logFields map[string]string, err error) {
+	logFields = make(map[string]string)
+	from, nonce, err := h.node.Application.CheckTx(tx)
+	logFields["account"] = from.String()
+	logFields["nonce"] = fmt.Sprintf("%d", nonce)
+	if err != nil {
+		return
 	}
-	if err := h.node.Angine.BroadcastTx(tx); err != nil {
-		return nil, err
+	if err = h.node.Angine.BroadcastTx(tx); err != nil {
+		return
 	}
 
 	hash := gtypes.Tx(tx).Hash()
-	return &gtypes.ResultBroadcastTx{TxHash: hexutil.Encode(hash), Code: 0}, nil
+	return &gtypes.ResultBroadcastTx{TxHash: hexutil.Encode(hash), Code: 0}, logFields, nil
 }
 
-func (h *rpcHandler) BroadcastTxCommit(tx []byte) (*gtypes.ResultBroadcastTxCommit, error) {
-	if err := h.node.Application.CheckTx(tx); err != nil {
-		return nil, err
+func (h *rpcHandler) BroadcastTxCommit(tx []byte) (result *gtypes.ResultBroadcastTxCommit, logFields map[string]string, err error) {
+	logFields = make(map[string]string)
+	from, nonce, err := h.node.Application.CheckTx(tx)
+	logFields["account"] = from.String()
+	logFields["nonce"] = fmt.Sprintf("%d", nonce)
+	if err != nil {
+		return
 	}
-	if err := h.node.Angine.BroadcastTxCommit(tx); err != nil {
-		return nil, err
+	if err = h.node.Angine.BroadcastTxCommit(tx); err != nil {
+		return
 	}
 
 	hash := gtypes.Tx(tx).Hash()
-	return &gtypes.ResultBroadcastTxCommit{TxHash: hexutil.Encode(hash), Code: 0}, nil
+	return &gtypes.ResultBroadcastTxCommit{TxHash: hexutil.Encode(hash), Code: 0}, logFields, nil
 }
 
 func (h *rpcHandler) QueryTx(query []byte) (*gtypes.ResultNumLimitTx, error) {

chain/types/types.go

@@ -37,7 +37,7 @@ type (
 
 	KVs []*KV
 
-	QueryType       = byte
+	QueryType = byte
 )
 
 const (
@@ -53,6 +53,7 @@ const (
 	QueryTypeContractByHeight QueryType = 10
 	QueryType_Key             QueryType = 11
 	QueryType_Key_Prefix      QueryType = 12
+	QueryType_Pending_Nonce   QueryType = 13
 )
 
-var KVTxType = []byte("kvTx-")
+var KVTxType = []byte("kvTx-")

gemmill/angine.go

@@ -150,7 +150,7 @@ func NewAngine(app types.Application, tune *Tunes) (angine *Angine, err error) {
 
 	logger, err := getLogger(conf)
 	if err != nil {
-		fmt.Println("failed to get logger: ", err)
+		err  = fmt.Errorf("failed to get logger: %v", err)
 		return nil, err
 	}
 	log.SetLog(logger)
@@ -163,13 +163,13 @@ func NewAngine(app types.Application, tune *Tunes) (angine *Angine, err error) {
 	crypto.NodeInit(crypto.CryptoType)
 	privValidator, err := types.LoadPrivValidator(conf.GetString("priv_validator_file"))
 	if err != nil {
-		fmt.Println("LoadPrivValidator error: ", err)
+		err  = fmt.Errorf("LoadPrivValidator error: %v", err)
 		return nil, err
 	}
 	refuseList = refuse_list.NewRefuseList(dbBackend, dbDir)
 	p2psw, err := prepareP2P(conf, genesis, privValidator, refuseList)
 	if err != nil {
-		fmt.Println("prepare p2p error: ", err)
+		err  = fmt.Errorf("prepare p2p error: %v", err)
 		return nil, err
 	}
 
@@ -222,16 +222,16 @@ func (a *Angine) OnRecvExchangeData(data *p2p.ExchangeData) error {
 			// TODO wait ...
 			return errors.New("no genesis file found in other node")
 		}
-		othGenesis, err := types.GenesisDocFromJSONRet(data.GenesisJSON)
+		otherGenesis, err := types.GenesisDocFromJSONRet(data.GenesisJSON)
 		if err != nil {
 			// TODO log err
-			fmt.Println("oth genesis err:", err)
+			log.Warn("other genesis err:", zap.Error(err))
 			return err
 		}
 		a.p2pSwitch.GetExchangeData().GenesisJSON = data.GenesisJSON
-		if err = a.buildState(othGenesis); err != nil {
+		if err = a.buildState(otherGenesis); err != nil {
 			// TODO log err
-			fmt.Println("build state err:", err)
+			log.Warn("build state err:",  zap.Error(err))
 			return err
 		}
 		if a.stateMachine == nil {
@@ -245,7 +245,7 @@ func (a *Angine) OnRecvExchangeData(data *p2p.ExchangeData) error {
 func (a *Angine) buildState(genesis *types.GenesisDoc) error {
 	stateM, err := getOrMakeState(a.conf, a.dbs["state"], genesis)
 	if err != nil {
-		fmt.Println("getOrMakeState error: ", err)
+		log.Warn("getOrMakeState error: ", zap.Error(err))
 		return err
 	}
 
@@ -572,7 +572,7 @@ func (e *Angine) Start() error {
 	}
 
 	if _, err := (*e.eventSwitch).Start(); err != nil {
-		fmt.Println("fail to start event switch, error: ", err)
+		log.Warn("fail to start event switch, error: ", zap.Error(err))
 		return err
 	}
 
@@ -1008,7 +1008,6 @@ func (ang *Angine) InitPlugins() {
 			if err != nil {
 				// querydb failure is something that we can bear with
 				log.Error("[QueryCachePlugin Init]", zap.Error(err))
-				fmt.Println(err)
 			}
 			params.StateDB = querydb
 			p.Init(params)

gemmill/mempool/mempool.go

@@ -64,6 +64,10 @@ func NewMempool(conf *viper.Viper) *Mempool {
 	return mempool
 }
 
+func (mem *Mempool) GetPendingMaxNonce(byt []byte) (uint64, error) {
+	return 0, errors.New("raft consense please use GetNonce() interface")
+}
+
 func (mem *Mempool) RegisterFilter(filter types.IFilter) {
 	mem.txFilters = append(mem.txFilters, filter)
 }

gemmill/modules/go-autofile/group.go

@@ -19,7 +19,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"log"
 	"os"
 	"path"
 	"path/filepath"
@@ -30,6 +29,8 @@ import (
 	"time"
 
 	gcmn "github.com/dappledger/AnnChain/gemmill/modules/go-common"
+	"github.com/dappledger/AnnChain/gemmill/modules/go-log"
+	"go.uber.org/zap"
 )
 
 /*
@@ -216,18 +217,18 @@ func (g *Group) checkTotalSizeLimit() {
 		}
 		if index == gInfo.MaxIndex {
 			// Special degenerate case, just do nothing.
-			log.Println("WARNING: Group's head " + g.Head.Path + "may grow without bound")
+			log.Warn("WARNING: Group's head " + g.Head.Path + "may grow without bound")
 			return
 		}
 		pathToRemove := filePathForIndex(g.Head.Path, index, gInfo.MaxIndex)
 		fileInfo, err := os.Stat(pathToRemove)
 		if err != nil {
-			log.Println("WARNING: Failed to fetch info for file @" + pathToRemove)
+			log.Warn("WARNING: Failed to fetch info for file @" + pathToRemove)
 			continue
 		}
 		err = os.Remove(pathToRemove)
 		if err != nil {
-			log.Println(err)
+			log.Warn("remove file err",zap.Error(err))
 			return
 		}
 		totalSize -= fileInfo.Size()
@@ -245,7 +246,7 @@ func (g *Group) RotateFile() {
 		err := os.Rename(g.Head.Path, dstPath)
 		if err != nil {
 			if strings.Contains(err.Error(), "The process cannot access the file because it is being used by another process") {
-				log.Printf("Rename old(%s) to new(%s) error:%s\n", g.Head.Path, dstPath, err.Error())
+				log.Infof("Rename old(%s) to new(%s) error:%s\n", g.Head.Path, dstPath, err.Error())
 				err = g.Head.closeFile()
 				if err != nil {
 					panic(err)
@@ -255,7 +256,7 @@ func (g *Group) RotateFile() {
 				panic(err)
 			}
 		} else {
-			log.Printf("Rename old(%s) to new(%s) OK\n", g.Head.Path, dstPath)
+			log.Warnf("Rename old(%s) to new(%s) OK\n", g.Head.Path, dstPath)
 			break
 		}
 	}