Merge pull request #748 from nspcc-dev/feature/splitnep5

core: store NEP5 balances separately from account

pkg/core/blockchain.go

@@ -29,7 +29,7 @@ import (
 // Tuning parameters.
 const (
 	headerBatchCount = 2000
-	version          = "0.0.7"
+	version          = "0.0.8"
 
 	// This one comes from C# code and it's different from the constant used
 	// when creating an asset with Neo.Asset.Create interop call. It looks
@@ -763,44 +763,46 @@ func (bc *Blockchain) processNEP5Transfer(cache *cachedDao, tx *transaction.Tran
 		Tx:        tx.Hash(),
 	}
 	if !fromAddr.Equals(util.Uint160{}) {
-		acc, err := cache.GetAccountStateOrNew(fromAddr)
+		balances, err := cache.GetNEP5Balances(fromAddr)
 		if err != nil {
 			return
 		}
-		bs := acc.NEP5Balances[sc]
+		bs := balances.Trackers[sc]
-		if bs == nil {
-			bs = new(state.NEP5Tracker)
-			acc.NEP5Balances[sc] = bs
-		}
 		bs.Balance -= amount
 		bs.LastUpdatedBlock = b.Index
-		if err := cache.PutAccountState(acc); err != nil {
+		balances.Trackers[sc] = bs
-			return
-		}
 
 		transfer.Amount = -amount
-		if err := cache.AppendNEP5Transfer(fromAddr, transfer); err != nil {
+		isBig, err := cache.AppendNEP5Transfer(fromAddr, balances.NextTransferBatch, transfer)
+		if err != nil {
+			return
+		}
+		if isBig {
+			balances.NextTransferBatch++
+		}
+		if err := cache.PutNEP5Balances(fromAddr, balances); err != nil {
 			return
 		}
 	}
 	if !toAddr.Equals(util.Uint160{}) {
-		acc, err := cache.GetAccountStateOrNew(toAddr)
+		balances, err := cache.GetNEP5Balances(toAddr)
 		if err != nil {
 			return
 		}
-		bs := acc.NEP5Balances[sc]
+		bs := balances.Trackers[sc]
-		if bs == nil {
-			bs = new(state.NEP5Tracker)
-			acc.NEP5Balances[sc] = bs
-		}
 		bs.Balance += amount
 		bs.LastUpdatedBlock = b.Index
-		if err := cache.PutAccountState(acc); err != nil {
+		balances.Trackers[sc] = bs
-			return
-		}
 
 		transfer.Amount = amount
-		if err := cache.AppendNEP5Transfer(toAddr, transfer); err != nil {
+		isBig, err := cache.AppendNEP5Transfer(toAddr, balances.NextTransferBatch, transfer)
+		if err != nil {
+			return
+		}
+		if isBig {
+			balances.NextTransferBatch++
+		}
+		if err := cache.PutNEP5Balances(toAddr, balances); err != nil {
 			return
 		}
 	}
@@ -808,11 +810,28 @@ func (bc *Blockchain) processNEP5Transfer(cache *cachedDao, tx *transaction.Tran
 
 // GetNEP5TransferLog returns NEP5 transfer log for the acc.
 func (bc *Blockchain) GetNEP5TransferLog(acc util.Uint160) *state.NEP5TransferLog {
-	lg, err := bc.dao.GetNEP5TransferLog(acc)
+	balances, err := bc.dao.GetNEP5Balances(acc)
 	if err != nil {
 		return nil
 	}
-	return lg
+	result := new(state.NEP5TransferLog)
+	for i := uint32(0); i <= balances.NextTransferBatch; i++ {
+		lg, err := bc.dao.GetNEP5TransferLog(acc, i)
+		if err != nil {
+			return nil
+		}
+		result.Raw = append(result.Raw, lg.Raw...)
+	}
+	return result
+}
+
+// GetNEP5Balances returns NEP5 balances for the acc.
+func (bc *Blockchain) GetNEP5Balances(acc util.Uint160) *state.NEP5Balances {
+	bs, err := bc.dao.GetNEP5Balances(acc)
+	if err != nil {
+		return nil
+	}
+	return bs
 }
 
 // LastBatch returns last persisted storage batch.

pkg/core/blockchainer.go

@@ -36,6 +36,7 @@ type Blockchainer interface {
 	GetAccountState(util.Uint160) *state.Account
 	GetAppExecResult(util.Uint256) (*state.AppExecResult, error)
 	GetNEP5TransferLog(util.Uint160) *state.NEP5TransferLog
+	GetNEP5Balances(util.Uint160) *state.NEP5Balances
 	GetValidators(txes ...*transaction.Transaction) ([]*keys.PublicKey, error)
 	GetScriptHashesForVerifying(*transaction.Transaction) ([]util.Uint160, error)
 	GetStorageItem(scripthash util.Uint160, key []byte) *state.StorageItem

pkg/core/cacheddao.go

@@ -14,6 +14,8 @@ type cachedDao struct {
 	accounts  map[util.Uint160]*state.Account
 	contracts map[util.Uint160]*state.Contract
 	unspents  map[util.Uint256]*state.UnspentCoin
+	balances  map[util.Uint160]*state.NEP5Balances
+	transfers map[util.Uint160]map[uint32]*state.NEP5TransferLog
 }
 
 // newCachedDao returns new cachedDao wrapping around given backing store.
@@ -21,7 +23,9 @@ func newCachedDao(backend storage.Store) *cachedDao {
 	accs := make(map[util.Uint160]*state.Account)
 	ctrs := make(map[util.Uint160]*state.Contract)
 	unspents := make(map[util.Uint256]*state.UnspentCoin)
-	return &cachedDao{*newDao(backend), accs, ctrs, unspents}
+	balances := make(map[util.Uint160]*state.NEP5Balances)
+	transfers := make(map[util.Uint160]map[uint32]*state.NEP5TransferLog)
+	return &cachedDao{*newDao(backend), accs, ctrs, unspents, balances, transfers}
 }
 
 // GetAccountStateOrNew retrieves Account from cache or underlying Store
@@ -85,6 +89,52 @@ func (cd *cachedDao) PutUnspentCoinState(hash util.Uint256, ucs *state.UnspentCo
 	return nil
 }
 
+// GetNEP5Balances retrieves NEP5Balances for the acc.
+func (cd *cachedDao) GetNEP5Balances(acc util.Uint160) (*state.NEP5Balances, error) {
+	if bs := cd.balances[acc]; bs != nil {
+		return bs, nil
+	}
+	return cd.dao.GetNEP5Balances(acc)
+}
+
+// PutNEP5Balances saves NEP5Balances for the acc.
+func (cd *cachedDao) PutNEP5Balances(acc util.Uint160, bs *state.NEP5Balances) error {
+	cd.balances[acc] = bs
+	return nil
+}
+
+// GetNEP5TransferLog retrieves NEP5TransferLog for the acc.
+func (cd *cachedDao) GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.NEP5TransferLog, error) {
+	ts := cd.transfers[acc]
+	if ts != nil && ts[index] != nil {
+		return ts[index], nil
+	}
+	return cd.dao.GetNEP5TransferLog(acc, index)
+}
+
+// PutNEP5TransferLog saves NEP5TransferLog for the acc.
+func (cd *cachedDao) PutNEP5TransferLog(acc util.Uint160, index uint32, bs *state.NEP5TransferLog) error {
+	ts := cd.transfers[acc]
+	if ts == nil {
+		ts = make(map[uint32]*state.NEP5TransferLog, 2)
+		cd.transfers[acc] = ts
+	}
+	ts[index] = bs
+	return nil
+}
+
+// AppendNEP5Transfer appends new transfer to a transfer event log.
+func (cd *cachedDao) AppendNEP5Transfer(acc util.Uint160, index uint32, tr *state.NEP5Transfer) (bool, error) {
+	lg, err := cd.GetNEP5TransferLog(acc, index)
+	if err != nil {
+		return false, err
+	}
+	if err := lg.Append(tr); err != nil {
+		return false, err
+	}
+	return lg.Size() >= nep5TransferBatchSize, cd.PutNEP5TransferLog(acc, index, lg)
+}
+
 // Persist flushes all the changes made into the (supposedly) persistent
 // underlying store.
 func (cd *cachedDao) Persist() (int, error) {
@@ -100,5 +150,19 @@ func (cd *cachedDao) Persist() (int, error) {
 			return 0, err
 		}
 	}
+	for acc, bs := range cd.balances {
+		err := cd.dao.PutNEP5Balances(acc, bs)
+		if err != nil {
+			return 0, err
+		}
+	}
+	for acc, ts := range cd.transfers {
+		for ind, lg := range ts {
+			err := cd.dao.PutNEP5TransferLog(acc, ind, lg)
+			if err != nil {
+				return 0, err
+			}
+		}
+	}
 	return cd.dao.Persist()
 }

pkg/core/dao.go

@@ -135,11 +135,42 @@ func (dao *dao) DeleteContractState(hash util.Uint160) error {
 
 // -- end contracts.
 
+// -- start nep5 balances.
+
+// GetNEP5Balances retrieves nep5 balances from the cache.
+func (dao *dao) GetNEP5Balances(acc util.Uint160) (*state.NEP5Balances, error) {
+	key := storage.AppendPrefix(storage.STNEP5Balances, acc.BytesBE())
+	bs := state.NewNEP5Balances()
+	err := dao.GetAndDecode(bs, key)
+	if err != nil && err != storage.ErrKeyNotFound {
+		return nil, err
+	}
+	return bs, nil
+}
+
+// GetNEP5Balances saves nep5 balances from the cache.
+func (dao *dao) PutNEP5Balances(acc util.Uint160, bs *state.NEP5Balances) error {
+	key := storage.AppendPrefix(storage.STNEP5Balances, acc.BytesBE())
+	return dao.Put(bs, key)
+}
+
+// -- end nep5 balances.
+
 // -- start transfer log.
 
+const nep5TransferBatchSize = 128
+
+func getNEP5TransferLogKey(acc util.Uint160, index uint32) []byte {
+	key := make([]byte, 1+util.Uint160Size+4)
+	key[0] = byte(storage.STNEP5Transfers)
+	copy(key[1:], acc.BytesBE())
+	binary.LittleEndian.PutUint32(key[util.Uint160Size:], index)
+	return key
+}
+
 // GetNEP5TransferLog retrieves transfer log from the cache.
-func (dao *dao) GetNEP5TransferLog(acc util.Uint160) (*state.NEP5TransferLog, error) {
+func (dao *dao) GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.NEP5TransferLog, error) {
-	key := storage.AppendPrefix(storage.STNEP5Transfers, acc.BytesBE())
+	key := getNEP5TransferLogKey(acc, index)
 	value, err := dao.store.Get(key)
 	if err != nil {
 		if err == storage.ErrKeyNotFound {
@@ -151,24 +182,25 @@ func (dao *dao) GetNEP5TransferLog(acc util.Uint160) (*state.NEP5TransferLog, er
 }
 
 // PutNEP5TransferLog saves given transfer log in the cache.
-func (dao *dao) PutNEP5TransferLog(acc util.Uint160, lg *state.NEP5TransferLog) error {
+func (dao *dao) PutNEP5TransferLog(acc util.Uint160, index uint32, lg *state.NEP5TransferLog) error {
-	key := storage.AppendPrefix(storage.STNEP5Transfers, acc.BytesBE())
+	key := getNEP5TransferLogKey(acc, index)
 	return dao.store.Put(key, lg.Raw)
 }
 
 // AppendNEP5Transfer appends a single NEP5 transfer to a log.
-func (dao *dao) AppendNEP5Transfer(acc util.Uint160, tr *state.NEP5Transfer) error {
+// First return value signalizes that log size has exceeded batch size.
-	lg, err := dao.GetNEP5TransferLog(acc)
+func (dao *dao) AppendNEP5Transfer(acc util.Uint160, index uint32, tr *state.NEP5Transfer) (bool, error) {
+	lg, err := dao.GetNEP5TransferLog(acc, index)
 	if err != nil {
 		if err != storage.ErrKeyNotFound {
-			return err
+			return false, err
 		}
 		lg = new(state.NEP5TransferLog)
 	}
 	if err := lg.Append(tr); err != nil {
-		return err
+		return false, err
 	}
-	return dao.PutNEP5TransferLog(acc, lg)
+	return lg.Size() >= nep5TransferBatchSize, dao.PutNEP5TransferLog(acc, index, lg)
 }
 
 // -- end transfer log.

pkg/core/state/account.go

@@ -35,9 +35,6 @@ type Account struct {
 	Votes      []*keys.PublicKey
 	Balances   map[util.Uint256][]UnspentBalance
 	Unclaimed  []UnclaimedBalance
-	// NEP5Balances is a map of the NEP5 contract hashes
-	// to the corresponding structures.
-	NEP5Balances map[util.Uint160]*NEP5Tracker
 }
 
 // NewAccount returns a new Account object.
@@ -49,8 +46,6 @@ func NewAccount(scriptHash util.Uint160) *Account {
 		Votes:      []*keys.PublicKey{},
 		Balances:   make(map[util.Uint256][]UnspentBalance),
 		Unclaimed:  []UnclaimedBalance{},
-
-		NEP5Balances: make(map[util.Uint160]*NEP5Tracker),
 	}
 }
 
@@ -75,16 +70,6 @@ func (s *Account) DecodeBinary(br *io.BinReader) {
 	}
 
 	br.ReadArray(&s.Unclaimed)
-
-	lenBalances = br.ReadVarUint()
-	s.NEP5Balances = make(map[util.Uint160]*NEP5Tracker, lenBalances)
-	for i := 0; i < int(lenBalances); i++ {
-		var key util.Uint160
-		var tr NEP5Tracker
-		br.ReadBytes(key[:])
-		tr.DecodeBinary(br)
-		s.NEP5Balances[key] = &tr
-	}
 }
 
 // EncodeBinary encodes Account to the given BinWriter.
@@ -104,12 +89,6 @@ func (s *Account) EncodeBinary(bw *io.BinWriter) {
 	}
 
 	bw.WriteArray(s.Unclaimed)
-
-	bw.WriteVarUint(uint64(len(s.NEP5Balances)))
-	for k, v := range s.NEP5Balances {
-		bw.WriteBytes(k[:])
-		v.EncodeBinary(bw)
-	}
 }
 
 // DecodeBinary implements io.Serializable interface.

pkg/core/state/nep5.go

@@ -41,6 +41,46 @@ type NEP5Transfer struct {
 	Tx util.Uint256
 }
 
+// NEP5Balances is a map of the NEP5 contract hashes
+// to the corresponding structures.
+type NEP5Balances struct {
+	Trackers map[util.Uint160]NEP5Tracker
+	// NextTransferBatch stores an index of the next transfer batch.
+	NextTransferBatch uint32
+}
+
+// NewNEP5Balances returns new NEP5Balances.
+func NewNEP5Balances() *NEP5Balances {
+	return &NEP5Balances{
+		Trackers: make(map[util.Uint160]NEP5Tracker),
+	}
+}
+
+// DecodeBinary implements io.Serializable interface.
+func (bs *NEP5Balances) DecodeBinary(r *io.BinReader) {
+	bs.NextTransferBatch = r.ReadU32LE()
+	lenBalances := r.ReadVarUint()
+	m := make(map[util.Uint160]NEP5Tracker, lenBalances)
+	for i := 0; i < int(lenBalances); i++ {
+		var key util.Uint160
+		var tr NEP5Tracker
+		r.ReadBytes(key[:])
+		tr.DecodeBinary(r)
+		m[key] = tr
+	}
+	bs.Trackers = m
+}
+
+// EncodeBinary implements io.Serializable interface.
+func (bs *NEP5Balances) EncodeBinary(w *io.BinWriter) {
+	w.WriteU32LE(bs.NextTransferBatch)
+	w.WriteVarUint(uint64(len(bs.Trackers)))
+	for k, v := range bs.Trackers {
+		w.WriteBytes(k[:])
+		v.EncodeBinary(w)
+	}
+}
+
 // Append appends single transfer to a log.
 func (lg *NEP5TransferLog) Append(tr *NEP5Transfer) error {
 	w := io.NewBufBinWriter()
@@ -70,6 +110,11 @@ func (lg *NEP5TransferLog) ForEach(f func(*NEP5Transfer) error) error {
 	return nil
 }
 
+// Size returns an amount of transfer written in log.
+func (lg *NEP5TransferLog) Size() int {
+	return len(lg.Raw) / NEP5TransferSize
+}
+
 // EncodeBinary implements io.Serializable interface.
 func (t *NEP5Tracker) EncodeBinary(w *io.BinWriter) {
 	w.WriteU64LE(uint64(t.Balance))

pkg/core/state/nep5_test.go

@@ -25,6 +25,8 @@ func TestNEP5TransferLog_Append(t *testing.T) {
 		require.NoError(t, lg.Append(tr))
 	}
 
+	require.Equal(t, len(expected), lg.Size())
+
 	i := 0
 	err := lg.ForEach(func(tr *NEP5Transfer) error {
 		require.Equal(t, expected[i], tr)

pkg/core/storage/store.go

@@ -18,6 +18,7 @@ const (
 	STContract        KeyPrefix = 0x50
 	STStorage         KeyPrefix = 0x70
 	STNEP5Transfers   KeyPrefix = 0x72
+	STNEP5Balances    KeyPrefix = 0x73
 	IXHeaderHashList  KeyPrefix = 0x80
 	IXValidatorsCount KeyPrefix = 0x90
 	SYSCurrentBlock   KeyPrefix = 0xc0

pkg/network/helper_test.go

@@ -94,6 +94,9 @@ func (chain testChain) GetAccountState(util.Uint160) *state.Account {
 func (chain testChain) GetNEP5TransferLog(util.Uint160) *state.NEP5TransferLog {
 	panic("TODO")
 }
+func (chain testChain) GetNEP5Balances(util.Uint160) *state.NEP5Balances {
+	panic("TODO")
+}
 func (chain testChain) GetValidators(...*transaction.Transaction) ([]*keys.PublicKey, error) {
 	panic("TODO")
 }

pkg/rpc/server/server.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"math"
-	"math/big"
 	"net/http"
 	"strconv"
 
@@ -23,9 +22,9 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/rpc/request"
 	"github.com/nspcc-dev/neo-go/pkg/rpc/response"
 	"github.com/nspcc-dev/neo-go/pkg/rpc/response/result"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"github.com/nspcc-dev/neo-go/pkg/vm/emit"
-	"github.com/nspcc-dev/neo-go/pkg/vm/opcode"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 )
@@ -424,11 +423,14 @@ func (s *Server) getNEP5Balances(ps request.Params) (interface{}, error) {
 		return nil, response.ErrInvalidParams
 	}
 
-	as := s.chain.GetAccountState(u)
+	as := s.chain.GetNEP5Balances(u)
-	bs := &result.NEP5Balances{Address: address.Uint160ToString(u)}
+	bs := &result.NEP5Balances{
+		Address:  address.Uint160ToString(u),
+		Balances: []result.NEP5Balance{},
+	}
 	if as != nil {
 		cache := make(map[util.Uint160]int64)
-		for h, bal := range as.NEP5Balances {
+		for h, bal := range as.Trackers {
 			dec, err := s.getDecimals(h, cache)
 			if err != nil {
 				continue
@@ -454,7 +456,11 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, error) {
 		return nil, response.ErrInvalidParams
 	}
 
-	bs := &result.NEP5Transfers{Address: address.Uint160ToString(u)}
+	bs := &result.NEP5Transfers{
+		Address:  address.Uint160ToString(u),
+		Received: []result.NEP5Transfer{},
+		Sent:     []result.NEP5Transfer{},
+	}
 	lg := s.chain.GetNEP5TransferLog(u)
 	cache := make(map[util.Uint160]int64)
 	err = lg.ForEach(func(tr *state.NEP5Transfer) error {
@@ -508,29 +514,33 @@ func (s *Server) getDecimals(h util.Uint160, cache map[util.Uint160]int64) (int6
 	if d, ok := cache[h]; ok {
 		return d, nil
 	}
-	w := io.NewBufBinWriter()
+	script, err := request.CreateFunctionInvocationScript(h, request.Params{
-	emit.Int(w.BinWriter, 0)
+		{
-	emit.Opcode(w.BinWriter, opcode.NEWARRAY)
+			Type:  request.StringT,
-	emit.String(w.BinWriter, "decimals")
+			Value: "decimals",
-	emit.AppCall(w.BinWriter, h, true)
+		},
-	v, _ := s.chain.GetTestVM()
+		{
-	v.LoadScript(w.Bytes())
+			Type:  request.ArrayT,
-	if err := v.Run(); err != nil {
+			Value: []request.Param{},
+		},
+	})
+	if err != nil {
 		return 0, err
 	}
-	res := v.PopResult()
+	res := s.runScriptInVM(script)
-	if res == nil {
+	if res == nil || res.State != "HALT" || len(res.Stack) == 0 {
+		return 0, errors.New("execution error")
+	}
+
+	var d int64
+	switch item := res.Stack[len(res.Stack)-1]; item.Type {
+	case smartcontract.IntegerType:
+		d = item.Value.(int64)
+	case smartcontract.ByteArrayType:
+		d = emit.BytesToInt(item.Value.([]byte)).Int64()
+	default:
 		return 0, errors.New("invalid result")
 	}
-	bi, ok := res.(*big.Int)
-	if !ok {
-		bs, ok := res.([]byte)
-		if !ok {
-			return 0, errors.New("invalid result")
-		}
-		bi = emit.BytesToInt(bs)
-	}
-	d := bi.Int64()
 	if d < 0 {
 		return 0, errors.New("negative decimals")
 	}

pkg/rpc/server/server_test.go

@@ -623,7 +623,7 @@ var rpcTestCases = map[string][]rpcTestCase{
 				require.True(t, ok)
 				assert.Equal(t, res.Available, util.Fixed8FromInt64(8))
 				assert.True(t, res.Unavailable > 0)
-				assert.Equal(t, res.Available + res.Unavailable, res.Unclaimed)
+				assert.Equal(t, res.Available+res.Unavailable, res.Unclaimed)
 			},
 		},
 	},