diff --git a/pkg/core/blockchain.go b/pkg/core/blockchain.go index a0aba8148..82b26fbd7 100644 --- a/pkg/core/blockchain.go +++ b/pkg/core/blockchain.go @@ -613,6 +613,9 @@ func (bc *Blockchain) storeBlock(block *block.Block) error { return err } + var pseudoSender util.Uint160 + var gasTotal, neoTotal util.Fixed8 + // Process TX inputs that are grouped by previous hash. for _, inputs := range transaction.GroupInputsByPrevHash(tx.Inputs) { prevHash := inputs[0].PrevHash @@ -620,7 +623,7 @@ func (bc *Blockchain) storeBlock(block *block.Block) error { if err != nil { return err } - for _, input := range inputs { + for i, input := range inputs { if len(unspent.States) <= int(input.PrevIndex) { return fmt.Errorf("bad input: %s/%d", input.PrevHash.StringLE(), input.PrevIndex) } @@ -630,6 +633,14 @@ func (bc *Blockchain) storeBlock(block *block.Block) error { unspent.States[input.PrevIndex].State |= state.CoinSpent unspent.States[input.PrevIndex].SpendHeight = block.Index prevTXOutput := &unspent.States[input.PrevIndex].Output + if i == 0 { + pseudoSender = prevTXOutput.ScriptHash + } + if prevTXOutput.AssetID.Equals(GoverningTokenID()) { + neoTotal += prevTXOutput.Amount + } else if prevTXOutput.AssetID.Equals(UtilityTokenID()) { + gasTotal += prevTXOutput.Amount + } account, err := cache.GetAccountStateOrNew(prevTXOutput.ScriptHash) if err != nil { return err @@ -679,6 +690,10 @@ func (bc *Blockchain) storeBlock(block *block.Block) error { } } + if err := bc.processTransfer(cache, pseudoSender, tx, block, neoTotal, gasTotal); err != nil { + return err + } + // Process the underlying type of the TX. switch t := tx.Data.(type) { case *transaction.RegisterTX: @@ -906,6 +921,87 @@ func (bc *Blockchain) storeBlock(block *block.Block) error { return nil } +func appendSingleTransfer(cache *dao.Cached, acc util.Uint160, tr *state.Transfer) error { + index, err := cache.GetNextTransferBatch(acc) + if err != nil { + return err + } + isBig, err := cache.AppendTransfer(acc, index, tr) + if err != nil { + return err + } + if isBig { + if err := cache.PutNextTransferBatch(acc, index+1); err != nil { + return err + } + } + return nil +} + +// processTransfer processes single UTXO transfer. Totals is a slice of neo (0) and gas (1) total transfer amount. +func (bc *Blockchain) processTransfer(cache *dao.Cached, from util.Uint160, tx *transaction.Transaction, b *block.Block, + neoTotal, gasTotal util.Fixed8) error { + + fromIndex, err := cache.GetNextTransferBatch(from) + if err != nil { + return err + } + for i := range tx.Outputs { + isGoverning := tx.Outputs[i].AssetID.Equals(GoverningTokenID()) + if !isGoverning && !tx.Outputs[i].AssetID.Equals(UtilityTokenID()) { + continue + } + if !from.Equals(tx.Outputs[i].ScriptHash) { + tr := &state.Transfer{ + IsGoverning: isGoverning, + From: from, + To: tx.Outputs[i].ScriptHash, + Amount: int64(tx.Outputs[i].Amount), + Block: b.Index, + Timestamp: b.Timestamp, + Tx: tx.Hash(), + } + isBig, err := cache.AppendTransfer(from, fromIndex, tr) + if err != nil { + return err + } else if isBig { + fromIndex++ + } + if err := appendSingleTransfer(cache, tx.Outputs[i].ScriptHash, tr); err != nil { + return err + } + } + if isGoverning { + neoTotal -= tx.Outputs[i].Amount + } else { + gasTotal -= tx.Outputs[i].Amount + } + + } + for i, amount := range []util.Fixed8{neoTotal, gasTotal} { + if amount > 0 { + tr := &state.Transfer{ + IsGoverning: i == 0, + From: from, + Amount: int64(amount), + Block: b.Index, + Timestamp: b.Timestamp, + Tx: tx.Hash(), + } + isBig, err := cache.AppendTransfer(from, fromIndex, tr) + if err != nil { + return err + } else if isBig { + fromIndex++ + } + if err := appendSingleTransfer(cache, util.Uint160{}, tr); err != nil { + return err + } + } + } + return cache.PutNextTransferBatch(from, fromIndex) +} + func parseUint160(addr []byte) util.Uint160 { if u, err := util.Uint160DecodeBytesBE(addr); err == nil { return u @@ -971,21 +1067,42 @@ func (bc *Blockchain) processNEP5Transfer(cache *dao.Cached, tx *transaction.Tra } } -// GetNEP5TransferLog returns NEP5 transfer log for the acc. -func (bc *Blockchain) GetNEP5TransferLog(acc util.Uint160) *state.NEP5TransferLog { +// ForEachTransfer executes f for each transfer in log. +func (bc *Blockchain) ForEachTransfer(acc util.Uint160, tr *state.Transfer, f func() error) error { + nb, err := bc.dao.GetNextTransferBatch(acc) + if err != nil { + return nil + } + for i := uint32(0); i <= nb; i++ { + lg, err := bc.dao.GetTransferLog(acc, i) + if err != nil { + return nil + } + err = lg.ForEach(state.TransferSize, tr, f) + if err != nil { + return err + } + } + return nil +} + +// ForEachNEP5Transfer executes f for each nep5 transfer in log. +func (bc *Blockchain) ForEachNEP5Transfer(acc util.Uint160, tr *state.NEP5Transfer, f func() error) error { balances, err := bc.dao.GetNEP5Balances(acc) if err != nil { return nil } - 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...) + err = lg.ForEach(state.NEP5TransferSize, tr, f) + if err != nil { + return err + } } - return result + return nil } // GetNEP5Balances returns NEP5 balances for the acc. diff --git a/pkg/core/blockchainer.go b/pkg/core/blockchainer.go index ea64853c9..03f0bd4e3 100644 --- a/pkg/core/blockchainer.go +++ b/pkg/core/blockchainer.go @@ -26,6 +26,8 @@ type Blockchainer interface { GetBlock(hash util.Uint256) (*block.Block, error) GetContractState(hash util.Uint160) *state.Contract GetEnrollments() ([]*state.Validator, error) + ForEachNEP5Transfer(util.Uint160, *state.NEP5Transfer, func() error) error + ForEachTransfer(util.Uint160, *state.Transfer, func() error) error GetHeaderHash(int) util.Uint256 GetHeader(hash util.Uint256) (*block.Header, error) CurrentHeaderHash() util.Uint256 @@ -36,7 +38,6 @@ type Blockchainer interface { GetAccountState(util.Uint160) *state.Account GetAppExecResult(util.Uint256) (*state.AppExecResult, error) GetNEP5Metadata(util.Uint160) (*state.NEP5Metadata, error) - GetNEP5TransferLog(util.Uint160) *state.NEP5TransferLog GetNEP5Balances(util.Uint160) *state.NEP5Balances GetValidators(txes ...*transaction.Transaction) ([]*keys.PublicKey, error) GetScriptHashesForVerifying(*transaction.Transaction) ([]util.Uint160, error) diff --git a/pkg/core/dao/cacheddao.go b/pkg/core/dao/cacheddao.go index 3a64b2b58..3309aa254 100644 --- a/pkg/core/dao/cacheddao.go +++ b/pkg/core/dao/cacheddao.go @@ -16,12 +16,14 @@ import ( // objects in the storeBlock(). type Cached struct { DAO - 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 - storage *itemCache + accounts map[util.Uint160]*state.Account + contracts map[util.Uint160]*state.Contract + unspents map[util.Uint256]*state.UnspentCoin + balances map[util.Uint160]*state.NEP5Balances + nep5transfers map[util.Uint160]map[uint32]*state.TransferLog + transfers map[util.Uint160]map[uint32]*state.TransferLog + nextBatch map[util.Uint160]uint32 + storage *itemCache dropNEP5Cache bool } @@ -32,7 +34,9 @@ func NewCached(d DAO) *Cached { ctrs := make(map[util.Uint160]*state.Contract) unspents := make(map[util.Uint256]*state.UnspentCoin) balances := make(map[util.Uint160]*state.NEP5Balances) - transfers := make(map[util.Uint160]map[uint32]*state.NEP5TransferLog) + nep5transfers := make(map[util.Uint160]map[uint32]*state.TransferLog) + transfers := make(map[util.Uint160]map[uint32]*state.TransferLog) + nextBatch := make(map[util.Uint160]uint32) st := newItemCache() dao := d.GetWrapped() if cd, ok := dao.(*Cached); ok { @@ -42,7 +46,18 @@ func NewCached(d DAO) *Cached { } } } - return &Cached{dao, accs, ctrs, unspents, balances, transfers, st, false} + return &Cached{ + DAO: dao, + accounts: accs, + contracts: ctrs, + unspents: unspents, + balances: balances, + nep5transfers: nep5transfers, + transfers: transfers, + nextBatch: nextBatch, + storage: st, + dropNEP5Cache: false, + } } // GetAccountStateOrNew retrieves Account from cache or underlying store @@ -106,6 +121,52 @@ func (cd *Cached) PutUnspentCoinState(hash util.Uint256, ucs *state.UnspentCoin) return nil } +// GetNextTransferBatch returns index for the transfer batch to write to. +func (cd *Cached) GetNextTransferBatch(acc util.Uint160) (uint32, error) { + if n, ok := cd.nextBatch[acc]; ok { + return n, nil + } + return cd.DAO.GetNextTransferBatch(acc) +} + +// PutNextTransferBatch sets index of the transfer batch to write to. +func (cd *Cached) PutNextTransferBatch(acc util.Uint160, num uint32) error { + cd.nextBatch[acc] = num + return nil +} + +// GetTransferLog retrieves TransferLog for the acc. +func (cd *Cached) GetTransferLog(acc util.Uint160, index uint32) (*state.TransferLog, error) { + ts := cd.transfers[acc] + if ts != nil && ts[index] != nil { + return ts[index], nil + } + return cd.DAO.GetTransferLog(acc, index) +} + +// PutTransferLog saves TransferLog for the acc. +func (cd *Cached) PutTransferLog(acc util.Uint160, index uint32, bs *state.TransferLog) error { + ts := cd.transfers[acc] + if ts == nil { + ts = make(map[uint32]*state.TransferLog, 2) + cd.transfers[acc] = ts + } + ts[index] = bs + return nil +} + +// AppendTransfer appends new transfer to a transfer event log. +func (cd *Cached) AppendTransfer(acc util.Uint160, index uint32, tr *state.Transfer) (bool, error) { + lg, err := cd.GetTransferLog(acc, index) + if err != nil { + return false, err + } + if err := lg.Append(tr); err != nil { + return false, err + } + return lg.Size() >= transferBatchSize, cd.PutTransferLog(acc, index, lg) +} + // GetNEP5Balances retrieves NEP5Balances for the acc. func (cd *Cached) GetNEP5Balances(acc util.Uint160) (*state.NEP5Balances, error) { if bs := cd.balances[acc]; bs != nil { @@ -120,21 +181,21 @@ func (cd *Cached) PutNEP5Balances(acc util.Uint160, bs *state.NEP5Balances) erro return nil } -// GetNEP5TransferLog retrieves NEP5TransferLog for the acc. -func (cd *Cached) GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.NEP5TransferLog, error) { - ts := cd.transfers[acc] +// GetNEP5TransferLog retrieves TransferLog for the acc. +func (cd *Cached) GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.TransferLog, error) { + ts := cd.nep5transfers[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 *Cached) PutNEP5TransferLog(acc util.Uint160, index uint32, bs *state.NEP5TransferLog) error { - ts := cd.transfers[acc] +// PutNEP5TransferLog saves TransferLog for the acc. +func (cd *Cached) PutNEP5TransferLog(acc util.Uint160, index uint32, bs *state.TransferLog) error { + ts := cd.nep5transfers[acc] if ts == nil { - ts = make(map[uint32]*state.NEP5TransferLog, 2) - cd.transfers[acc] = ts + ts = make(map[uint32]*state.TransferLog, 2) + cd.nep5transfers[acc] = ts } ts[index] = bs return nil @@ -265,7 +326,7 @@ func (cd *Cached) Persist() (int, error) { } buf.Reset() } - for acc, ts := range cd.transfers { + for acc, ts := range cd.nep5transfers { for ind, lg := range ts { err := cd.DAO.PutNEP5TransferLog(acc, ind, lg) if err != nil { @@ -273,6 +334,20 @@ func (cd *Cached) Persist() (int, error) { } } } + for acc, ts := range cd.transfers { + for ind, lg := range ts { + err := cd.DAO.PutTransferLog(acc, ind, lg) + if err != nil { + return 0, err + } + } + } + for acc, nb := range cd.nextBatch { + err := cd.DAO.PutNextTransferBatch(acc, nb) + if err != nil { + return 0, err + } + } return cd.DAO.Persist() } @@ -283,7 +358,9 @@ func (cd *Cached) GetWrapped() DAO { cd.contracts, cd.unspents, cd.balances, + cd.nep5transfers, cd.transfers, + cd.nextBatch, cd.storage, false, } diff --git a/pkg/core/dao/dao.go b/pkg/core/dao/dao.go index 726b036b8..550d873ef 100644 --- a/pkg/core/dao/dao.go +++ b/pkg/core/dao/dao.go @@ -19,6 +19,7 @@ import ( // DAO is a data access object. type DAO interface { AppendNEP5Transfer(acc util.Uint160, index uint32, tr *state.NEP5Transfer) (bool, error) + AppendTransfer(acc util.Uint160, index uint32, tr *state.Transfer) (bool, error) DeleteContractState(hash util.Uint160) error DeleteStorageItem(scripthash util.Uint160, key []byte) error DeleteValidatorState(vs *state.Validator) error @@ -36,12 +37,14 @@ type DAO interface { GetHeaderHashes() ([]util.Uint256, error) GetNEP5Balances(acc util.Uint160) (*state.NEP5Balances, error) GetNEP5Metadata(h util.Uint160) (*state.NEP5Metadata, error) - GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.NEP5TransferLog, error) + GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.TransferLog, error) + GetNextTransferBatch(acc util.Uint160) (uint32, error) GetStateRoot(height uint32) (*state.MPTRootState, error) PutStateRoot(root *state.MPTRootState) error GetStorageItem(scripthash util.Uint160, key []byte) *state.StorageItem GetStorageItems(hash util.Uint160, prefix []byte) ([]StorageItemWithKey, error) GetTransaction(hash util.Uint256) (*transaction.Transaction, uint32, error) + GetTransferLog(acc util.Uint160, index uint32) (*state.TransferLog, error) GetUnspentCoinState(hash util.Uint256) (*state.UnspentCoin, error) GetValidatorState(publicKey *keys.PublicKey) (*state.Validator, error) GetValidatorStateOrNew(publicKey *keys.PublicKey) (*state.Validator, error) @@ -60,8 +63,10 @@ type DAO interface { PutCurrentHeader(hashAndIndex []byte) error PutNEP5Balances(acc util.Uint160, bs *state.NEP5Balances) error PutNEP5Metadata(h util.Uint160, meta *state.NEP5Metadata) error - PutNEP5TransferLog(acc util.Uint160, index uint32, lg *state.NEP5TransferLog) error + PutNEP5TransferLog(acc util.Uint160, index uint32, lg *state.TransferLog) error + PutNextTransferBatch(acc util.Uint160, num uint32) error PutStorageItem(scripthash util.Uint160, key []byte, si *state.StorageItem) error + PutTransferLog(acc util.Uint160, index uint32, lg *state.TransferLog) error PutUnspentCoinState(hash util.Uint256, ucs *state.UnspentCoin) error PutValidatorState(vs *state.Validator) error PutValidatorsCount(vc *state.ValidatorsCount) error @@ -262,7 +267,16 @@ func (dao *Simple) putNEP5Balances(acc util.Uint160, bs *state.NEP5Balances, buf // -- start transfer log. -const nep5TransferBatchSize = 128 +const nep5TransferBatchSize = 128 * state.NEP5TransferSize +const transferBatchSize = 128 * state.TransferSize + +func getTransferLogKey(acc util.Uint160, index uint32) []byte { + key := make([]byte, 1+util.Uint160Size+4) + key[0] = byte(storage.STTransfers) + copy(key[1:], acc.BytesBE()) + binary.LittleEndian.PutUint32(key[util.Uint160Size:], index) + return key +} func getNEP5TransferLogKey(acc util.Uint160, index uint32) []byte { key := make([]byte, 1+util.Uint160Size+4) @@ -272,21 +286,77 @@ func getNEP5TransferLogKey(acc util.Uint160, index uint32) []byte { return key } +// GetNextTransferBatch returns index for the transfer batch to write to. +func (dao *Simple) GetNextTransferBatch(acc util.Uint160) (uint32, error) { + key := storage.AppendPrefix(storage.STTransfers, acc.BytesBE()) + val, err := dao.Store.Get(key) + if err != nil { + if err != storage.ErrKeyNotFound { + return 0, err + } + return 0, nil + } + return binary.LittleEndian.Uint32(val), nil +} + +// PutNextTransferBatch sets index of the transfer batch to write to. +func (dao *Simple) PutNextTransferBatch(acc util.Uint160, num uint32) error { + key := storage.AppendPrefix(storage.STTransfers, acc.BytesBE()) + val := make([]byte, 4) + binary.LittleEndian.PutUint32(val, num) + return dao.Store.Put(key, val) +} + +// GetTransferLog retrieves transfer log from the cache. +func (dao *Simple) GetTransferLog(acc util.Uint160, index uint32) (*state.TransferLog, error) { + key := getTransferLogKey(acc, index) + value, err := dao.Store.Get(key) + if err != nil { + if err == storage.ErrKeyNotFound { + return new(state.TransferLog), nil + } + return nil, err + } + return &state.TransferLog{Raw: value}, nil +} + +// PutTransferLog saves given transfer log in the cache. +func (dao *Simple) PutTransferLog(acc util.Uint160, index uint32, lg *state.TransferLog) error { + key := getTransferLogKey(acc, index) + return dao.Store.Put(key, lg.Raw) +} + +// AppendTransfer appends a single transfer to a log. +// First return value signalizes that log size has exceeded batch size. +func (dao *Simple) AppendTransfer(acc util.Uint160, index uint32, tr *state.Transfer) (bool, error) { + lg, err := dao.GetTransferLog(acc, index) + if err != nil { + if err != storage.ErrKeyNotFound { + return false, err + } + lg = new(state.TransferLog) + } + if err := lg.Append(tr); err != nil { + return false, err + } + return lg.Size() >= transferBatchSize, dao.PutTransferLog(acc, index, lg) +} + // GetNEP5TransferLog retrieves transfer log from the cache. -func (dao *Simple) GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.NEP5TransferLog, error) { +func (dao *Simple) GetNEP5TransferLog(acc util.Uint160, index uint32) (*state.TransferLog, error) { key := getNEP5TransferLogKey(acc, index) value, err := dao.Store.Get(key) if err != nil { if err == storage.ErrKeyNotFound { - return new(state.NEP5TransferLog), nil + return new(state.TransferLog), nil } return nil, err } - return &state.NEP5TransferLog{Raw: value}, nil + return &state.TransferLog{Raw: value}, nil } // PutNEP5TransferLog saves given transfer log in the cache. -func (dao *Simple) PutNEP5TransferLog(acc util.Uint160, index uint32, lg *state.NEP5TransferLog) error { +func (dao *Simple) PutNEP5TransferLog(acc util.Uint160, index uint32, lg *state.TransferLog) error { key := getNEP5TransferLogKey(acc, index) return dao.Store.Put(key, lg.Raw) } @@ -299,7 +369,7 @@ func (dao *Simple) AppendNEP5Transfer(acc util.Uint160, index uint32, tr *state. if err != storage.ErrKeyNotFound { return false, err } - lg = new(state.NEP5TransferLog) + lg = new(state.TransferLog) } if err := lg.Append(tr); err != nil { return false, err diff --git a/pkg/core/state/nep5.go b/pkg/core/state/nep5.go index a3e510f40..0ef8543f5 100644 --- a/pkg/core/state/nep5.go +++ b/pkg/core/state/nep5.go @@ -14,8 +14,8 @@ type NEP5Tracker struct { LastUpdatedBlock uint32 } -// NEP5TransferLog is a log of NEP5 token transfers for the specific command. -type NEP5TransferLog struct { +// TransferLog is a log of NEP5 token transfers for the specific command. +type TransferLog struct { Raw []byte } @@ -99,7 +99,7 @@ func (bs *NEP5Metadata) EncodeBinary(w *io.BinWriter) { } // Append appends single transfer to a log. -func (lg *NEP5TransferLog) Append(tr *NEP5Transfer) error { +func (lg *TransferLog) Append(tr io.Serializable) error { w := io.NewBufBinWriter() tr.EncodeBinary(w.BinWriter) if w.Err != nil { @@ -110,26 +110,25 @@ func (lg *NEP5TransferLog) Append(tr *NEP5Transfer) error { } // ForEach iterates over transfer log returning on first error. -func (lg *NEP5TransferLog) ForEach(f func(*NEP5Transfer) error) error { +func (lg *TransferLog) ForEach(size int, tr io.Serializable, f func() error) error { if lg == nil { return nil } - tr := new(NEP5Transfer) - for i := 0; i < len(lg.Raw); i += NEP5TransferSize { - r := io.NewBinReaderFromBuf(lg.Raw[i : i+NEP5TransferSize]) + for i := 0; i < len(lg.Raw); i += size { + r := io.NewBinReaderFromBuf(lg.Raw[i : i+size]) tr.DecodeBinary(r) if r.Err != nil { return r.Err - } else if err := f(tr); err != nil { + } else if err := f(); err != nil { return nil } } return nil } -// Size returns an amount of transfer written in log. -func (lg *NEP5TransferLog) Size() int { - return len(lg.Raw) / NEP5TransferSize +// Size returns an amount of transfer written in log provided size of a single transfer. +func (lg *TransferLog) Size() int { + return len(lg.Raw) } // EncodeBinary implements io.Serializable interface. diff --git a/pkg/core/state/nep5_test.go b/pkg/core/state/nep5_test.go index 543d05819..90fb01e16 100644 --- a/pkg/core/state/nep5_test.go +++ b/pkg/core/state/nep5_test.go @@ -21,15 +21,16 @@ func TestNEP5TransferLog_Append(t *testing.T) { randomTransfer(r), } - lg := new(NEP5TransferLog) + lg := new(TransferLog) for _, tr := range expected { require.NoError(t, lg.Append(tr)) } - require.Equal(t, len(expected), lg.Size()) + require.Equal(t, len(expected), lg.Size()/NEP5TransferSize) i := 0 - err := lg.ForEach(func(tr *NEP5Transfer) error { + tr := new(NEP5Transfer) + err := lg.ForEach(NEP5TransferSize, tr, func() error { require.Equal(t, expected[i], tr) i++ return nil @@ -77,3 +78,7 @@ func randomTransfer(r *rand.Rand) *NEP5Transfer { Tx: random.Uint256(), } } + +func TestTransfer_Size(t *testing.T) { + require.Equal(t, TransferSize, io.GetVarSize(new(Transfer))) +} diff --git a/pkg/core/state/transfer_log.go b/pkg/core/state/transfer_log.go new file mode 100644 index 000000000..c89d638c2 --- /dev/null +++ b/pkg/core/state/transfer_log.go @@ -0,0 +1,51 @@ +package state + +import ( + "github.com/nspcc-dev/neo-go/pkg/io" + "github.com/nspcc-dev/neo-go/pkg/util" +) + +// TransferSize is a size of a marshaled Transfer struct in bytes. +const TransferSize = 1 + util.Uint160Size*2 + 8 + 4 + 4 + util.Uint256Size + +// Transfer represents a single Transfer event. +type Transfer struct { + // IsGoverning is true iff transfer is for neo token. + IsGoverning bool + // Address is the address of the sender. + From util.Uint160 + // To is the address of the receiver. + To util.Uint160 + // Amount is the amount of tokens transferred. + // It is negative when tokens are sent and positive if they are received. + Amount int64 + // Block is a number of block when the event occured. + Block uint32 + // Timestamp is the timestamp of the block where transfer occured. + Timestamp uint32 + // Tx is a hash the transaction. + Tx util.Uint256 +} + +// EncodeBinary implements io.Serializable interface. +// Note: change TransferSize constant when changing this function. +func (t *Transfer) EncodeBinary(w *io.BinWriter) { + w.WriteBytes(t.Tx[:]) + w.WriteBytes(t.From[:]) + w.WriteBytes(t.To[:]) + w.WriteU32LE(t.Block) + w.WriteU32LE(t.Timestamp) + w.WriteU64LE(uint64(t.Amount)) + w.WriteBool(t.IsGoverning) +} + +// DecodeBinary implements io.Serializable interface. +func (t *Transfer) DecodeBinary(r *io.BinReader) { + r.ReadBytes(t.Tx[:]) + r.ReadBytes(t.From[:]) + r.ReadBytes(t.To[:]) + t.Block = r.ReadU32LE() + t.Timestamp = r.ReadU32LE() + t.Amount = int64(r.ReadU64LE()) + t.IsGoverning = r.ReadBool() +} diff --git a/pkg/core/storage/store.go b/pkg/core/storage/store.go index 4f8324f14..ff09321c8 100644 --- a/pkg/core/storage/store.go +++ b/pkg/core/storage/store.go @@ -13,6 +13,7 @@ const ( STAccount KeyPrefix = 0x40 STCoin KeyPrefix = 0x44 STSpentCoin KeyPrefix = 0x45 + STTransfers KeyPrefix = 0x47 STValidator KeyPrefix = 0x48 STAsset KeyPrefix = 0x4c STNotification KeyPrefix = 0x4d diff --git a/pkg/network/helper_test.go b/pkg/network/helper_test.go index 2789364df..93aa36ccf 100644 --- a/pkg/network/helper_test.go +++ b/pkg/network/helper_test.go @@ -96,7 +96,7 @@ func (chain testChain) GetAccountState(util.Uint160) *state.Account { func (chain testChain) GetNEP5Metadata(util.Uint160) (*state.NEP5Metadata, error) { panic("TODO") } -func (chain testChain) GetNEP5TransferLog(util.Uint160) *state.NEP5TransferLog { +func (chain testChain) ForEachNEP5Transfer(util.Uint160, *state.NEP5Transfer, func() error) error { panic("TODO") } func (chain testChain) GetNEP5Balances(util.Uint160) *state.NEP5Balances { @@ -108,6 +108,9 @@ func (chain testChain) GetValidators(...*transaction.Transaction) ([]*keys.Publi func (chain testChain) GetEnrollments() ([]*state.Validator, error) { panic("TODO") } +func (chain testChain) ForEachTransfer(util.Uint160, *state.Transfer, func() error) error { + panic("TODO") +} func (chain testChain) GetScriptHashesForVerifying(*transaction.Transaction) ([]util.Uint160, error) { panic("TODO") } diff --git a/pkg/rpc/response/result/utxo.go b/pkg/rpc/response/result/utxo.go new file mode 100644 index 000000000..1bd2bb437 --- /dev/null +++ b/pkg/rpc/response/result/utxo.go @@ -0,0 +1,27 @@ +package result + +import "github.com/nspcc-dev/neo-go/pkg/util" + +// UTXO represents single output for a single asset. +type UTXO struct { + Index uint32 `json:"block_index"` + Timestamp uint32 `json:"timestamp"` + TxHash util.Uint256 `json:"txid"` + Address util.Uint160 `json:"transfer_address"` + Amount int64 `json:"amount,string"` +} + +// AssetUTXO represents UTXO for a specific asset. +type AssetUTXO struct { + AssetHash util.Uint256 `json:"asset_hash"` + AssetName string `json:"asset"` + TotalAmount int64 `json:"total_amount,string"` + Transactions []UTXO `json:"transactions"` +} + +// GetUTXO is a result of the `getutxotransfers` RPC. +type GetUTXO struct { + Address string `json:"address"` + Sent []AssetUTXO `json:"sent"` + Received []AssetUTXO `json:"received"` +} diff --git a/pkg/rpc/server/server.go b/pkg/rpc/server/server.go index 752652280..07318983c 100644 --- a/pkg/rpc/server/server.go +++ b/pkg/rpc/server/server.go @@ -8,6 +8,7 @@ import ( "net" "net/http" "strconv" + "strings" "sync" "time" @@ -103,6 +104,7 @@ var rpcHandlers = map[string]func(*Server, request.Params) (interface{}, *respon "getunspents": (*Server).getUnspents, "getvalidators": (*Server).getValidators, "getversion": (*Server).getVersion, + "getutxotransfers": (*Server).getUTXOTransfers, "invoke": (*Server).invoke, "invokefunction": (*Server).invokeFunction, "invokescript": (*Server).invokescript, @@ -447,6 +449,133 @@ func (s *Server) getVersion(_ request.Params) (interface{}, *response.Error) { }, nil } +func getTimestamps(p1, p2 *request.Param) (uint32, uint32, error) { + var start, end uint32 + if p1 != nil { + val, err := p1.GetInt() + if err != nil { + return 0, 0, err + } + start = uint32(val) + } + if p2 != nil { + val, err := p2.GetInt() + if err != nil { + return 0, 0, err + } + end = uint32(val) + } + return start, end, nil +} + +func getAssetMaps(name string) (map[util.Uint256]*result.AssetUTXO, map[util.Uint256]*result.AssetUTXO, error) { + sent := make(map[util.Uint256]*result.AssetUTXO) + recv := make(map[util.Uint256]*result.AssetUTXO) + name = strings.ToLower(name) + switch name { + case "neo", "gas", "": + default: + return nil, nil, errors.New("invalid asset") + } + if name == "neo" || name == "" { + sent[core.GoverningTokenID()] = &result.AssetUTXO{ + AssetHash: core.GoverningTokenID(), + AssetName: "NEO", + Transactions: []result.UTXO{}, + } + recv[core.GoverningTokenID()] = &result.AssetUTXO{ + AssetHash: core.GoverningTokenID(), + AssetName: "NEO", + Transactions: []result.UTXO{}, + } + } + if name == "gas" || name == "" { + sent[core.UtilityTokenID()] = &result.AssetUTXO{ + AssetHash: core.UtilityTokenID(), + AssetName: "GAS", + Transactions: []result.UTXO{}, + } + recv[core.UtilityTokenID()] = &result.AssetUTXO{ + AssetHash: core.UtilityTokenID(), + AssetName: "GAS", + Transactions: []result.UTXO{}, + } + } + return sent, recv, nil +} + +func (s *Server) getUTXOTransfers(ps request.Params) (interface{}, *response.Error) { + addr, err := ps.Value(0).GetUint160FromAddressOrHex() + if err != nil { + return nil, response.NewInvalidParamsError("", err) + } + + index := 1 + assetName, err := ps.Value(index).GetString() + if err == nil { + index++ + } + + start, end, err := getTimestamps(ps.Value(index), ps.Value(index+1)) + if err != nil { + return nil, response.NewInvalidParamsError("", err) + } + + sent, recv, err := getAssetMaps(assetName) + if err != nil { + return nil, response.NewInvalidParamsError("", err) + } + tr := new(state.Transfer) + err = s.chain.ForEachTransfer(addr, tr, func() error { + if tr.Timestamp < start || end != 0 && tr.Timestamp > end { + return nil + } + assetID := core.GoverningTokenID() + if !tr.IsGoverning { + assetID = core.UtilityTokenID() + } + a, ok := sent[assetID] + if ok && tr.From.Equals(addr) && !tr.To.Equals(addr) { + a.Transactions = append(a.Transactions, result.UTXO{ + Index: tr.Block, + Timestamp: tr.Timestamp, + TxHash: tr.Tx, + Address: tr.To, + Amount: tr.Amount, + }) + a.TotalAmount += tr.Amount + } + a, ok = recv[assetID] + if ok && tr.To.Equals(addr) && !tr.From.Equals(addr) { + a.Transactions = append(a.Transactions, result.UTXO{ + Index: tr.Block, + Timestamp: tr.Timestamp, + TxHash: tr.Tx, + Address: tr.From, + Amount: tr.Amount, + }) + a.TotalAmount += tr.Amount + } + return nil + }) + if err != nil { + return nil, response.NewInternalServerError("", err) + } + + res := &result.GetUTXO{ + Address: address.Uint160ToString(addr), + Sent: []result.AssetUTXO{}, + Received: []result.AssetUTXO{}, + } + for _, a := range sent { + res.Sent = append(res.Sent, *a) + } + for _, a := range recv { + res.Received = append(res.Received, *a) + } + return res, nil +} + func (s *Server) getPeers(_ request.Params) (interface{}, *response.Error) { peers := result.NewGetPeers() peers.AddUnconnected(s.coreServer.UnconnectedPeers()) @@ -597,8 +726,8 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err Received: []result.NEP5Transfer{}, Sent: []result.NEP5Transfer{}, } - lg := s.chain.GetNEP5TransferLog(u) - err = lg.ForEach(func(tr *state.NEP5Transfer) error { + tr := new(state.NEP5Transfer) + err = s.chain.ForEachNEP5Transfer(u, tr, func() error { transfer := result.NEP5Transfer{ Timestamp: tr.Timestamp, Asset: tr.Asset, diff --git a/pkg/rpc/server/server_test.go b/pkg/rpc/server/server_test.go index e9d130558..458a75603 100644 --- a/pkg/rpc/server/server_test.go +++ b/pkg/rpc/server/server_test.go @@ -290,6 +290,28 @@ var rpcTestCases = map[string][]rpcTestCase{ fail: true, }, }, + "getutxotransfers": { + { + name: "invalid address", + params: `["notanaddress"]`, + fail: true, + }, + { + name: "invalid asset", + params: `["AKkkumHbBipZ46UMZJoFynJMXzSRnBvKcs", "notanasset"]`, + fail: true, + }, + { + name: "invalid start timestamp", + params: `["AKkkumHbBipZ46UMZJoFynJMXzSRnBvKcs", "neo", "notanumber"]`, + fail: true, + }, + { + name: "invalid end timestamp", + params: `["AKkkumHbBipZ46UMZJoFynJMXzSRnBvKcs", "neo", 123, "notanumber"]`, + fail: true, + }, + }, "getassetstate": { { name: "positive", @@ -1104,6 +1126,39 @@ func testRPCProtocol(t *testing.T, doRPCCall func(string, string, *testing.T) [] assert.ElementsMatch(t, expected, actual) }) + + t.Run("getutxotransfers", func(t *testing.T) { + testGetUTXO := func(t *testing.T, asset string, start, stop int) { + ps := []string{`"AKkkumHbBipZ46UMZJoFynJMXzSRnBvKcs"`} + if asset != "" { + ps = append(ps, fmt.Sprintf("%q", asset)) + } + if start != 0 { + if start > int(e.chain.HeaderHeight()) { + ps = append(ps, strconv.Itoa(int(time.Now().Unix()))) + } else { + b, err := e.chain.GetHeader(e.chain.GetHeaderHash(start)) + require.NoError(t, err) + ps = append(ps, strconv.Itoa(int(b.Timestamp))) + } + if stop != 0 { + b, err := e.chain.GetHeader(e.chain.GetHeaderHash(stop)) + require.NoError(t, err) + ps = append(ps, strconv.Itoa(int(b.Timestamp))) + } + } + p := strings.Join(ps, ", ") + rpc := fmt.Sprintf(`{"jsonrpc": "2.0", "id": 1, "method": "getutxotransfers", "params": [%s]}`, p) + body := doRPCCall(rpc, httpSrv.URL, t) + res := checkErrGetResult(t, body, false) + actual := new(result.GetUTXO) + require.NoError(t, json.Unmarshal(res, actual)) + checkTransfers(t, e, actual, asset, start, stop) + } + t.Run("RestrictByAsset", func(t *testing.T) { testGetUTXO(t, "neo", 0, 0) }) + t.Run("TooBigStart", func(t *testing.T) { testGetUTXO(t, "", 300, 0) }) + t.Run("RestrictAll", func(t *testing.T) { testGetUTXO(t, "", 202, 203) }) + }) } func (tc rpcTestCase) getResultPair(e *executor) (expected interface{}, res interface{}) { @@ -1190,3 +1245,45 @@ func checkNep5Transfers(t *testing.T, e *executor, acc interface{}) { require.Equal(t, "AWLYWXB8C9Lt1nHdDZJnC5cpYJjgRDLk17", res.Sent[0].Address) require.Equal(t, uint32(0), res.Sent[0].NotifyIndex) } + +func checkTransfers(t *testing.T, e *executor, acc interface{}, asset string, start, stop int) { + res := acc.(*result.GetUTXO) + require.Equal(t, res.Address, "AKkkumHbBipZ46UMZJoFynJMXzSRnBvKcs") + + // transfer from multisig address to us + u := getUTXOForBlock(res, false, asset, 1) + if start <= 1 && (stop == 0 || stop >= 1) && (asset == "neo" || asset == "") { + require.NotNil(t, u) + require.Equal(t, "be48d3a3f5d10013ab9ffee489706078714f1ea2", u.Address.StringBE()) + require.EqualValues(t, int64(util.Fixed8FromInt64(99999000)), u.Amount) + } else { + require.Nil(t, u) + } + + // transfer from us to another validator + u = getUTXOForBlock(res, true, asset, 206) + if start <= 206 && (stop == 0 || stop >= 206) && (asset == "neo" || asset == "") { + require.NotNil(t, u) + require.Equal(t, "9fbf833320ef6bc52ddee1fe6f5793b42e9b307e", u.Address.StringBE()) + require.EqualValues(t, int64(util.Fixed8FromInt64(1000)), u.Amount) + } else { + require.Nil(t, u) + } +} + +func getUTXOForBlock(res *result.GetUTXO, sent bool, asset string, b uint32) *result.UTXO { + arr := res.Received + if sent { + arr = res.Sent + } + for i := range arr { + if arr[i].AssetName == strings.ToUpper(asset) { + for j := range arr[i].Transactions { + if b == arr[i].Transactions[j].Index { + return &arr[i].Transactions[j] + } + } + } + } + return nil +}