core/rpc: add continue flag to iterating functions

Most of the time we don't need to get all transfers from the DB and
deserialize them.

pkg/core/blockchain.go

@@ -1026,7 +1026,7 @@ func (bc *Blockchain) processNEP5Transfer(cache *dao.Cached, tx *transaction.Tra
 }
 
 // ForEachTransfer executes f for each transfer in log.
-func (bc *Blockchain) ForEachTransfer(acc util.Uint160, tr *state.Transfer, f func() error) error {
+func (bc *Blockchain) ForEachTransfer(acc util.Uint160, tr *state.Transfer, f func() (bool, error)) error {
 	nb, err := bc.dao.GetNextTransferBatch(acc)
 	if err != nil {
 		return nil
@@ -1036,16 +1036,19 @@ func (bc *Blockchain) ForEachTransfer(acc util.Uint160, tr *state.Transfer, f fu
 		if err != nil {
 			return nil
 		}
-		err = lg.ForEach(state.TransferSize, tr, f)
+		cont, err := lg.ForEach(state.TransferSize, tr, f)
 		if err != nil {
 			return err
 		}
+		if !cont {
+			break
+		}
 	}
 	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 {
+func (bc *Blockchain) ForEachNEP5Transfer(acc util.Uint160, tr *state.NEP5Transfer, f func() (bool, error)) error {
 	balances, err := bc.dao.GetNEP5Balances(acc)
 	if err != nil {
 		return nil
@@ -1055,10 +1058,13 @@ func (bc *Blockchain) ForEachNEP5Transfer(acc util.Uint160, tr *state.NEP5Transf
 		if err != nil {
 			return nil
 		}
-		err = lg.ForEach(state.NEP5TransferSize, tr, f)
+		cont, err := lg.ForEach(state.NEP5TransferSize, tr, f)
 		if err != nil {
 			return err
 		}
+		if !cont {
+			break
+		}
 	}
 	return nil
 }

pkg/core/blockchainer.go

@@ -26,8 +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
+	ForEachNEP5Transfer(util.Uint160, *state.NEP5Transfer, func() (bool, error)) error
-	ForEachTransfer(util.Uint160, *state.Transfer, func() error) error
+	ForEachTransfer(util.Uint160, *state.Transfer, func() (bool, error)) error
 	GetHeaderHash(int) util.Uint256
 	GetHeader(hash util.Uint256) (*block.Header, error)
 	CurrentHeaderHash() util.Uint256

pkg/core/state/nep5.go

@@ -110,20 +110,25 @@ func (lg *TransferLog) Append(tr io.Serializable) error {
 }
 
 // ForEach iterates over transfer log returning on first error.
-func (lg *TransferLog) ForEach(size int, tr io.Serializable, f func() error) error {
+func (lg *TransferLog) ForEach(size int, tr io.Serializable, f func() (bool, error)) (bool, error) {
 	if lg == nil {
-		return nil
+		return true, nil
 	}
 	for i := len(lg.Raw); i > 0; i -= size {
 		r := io.NewBinReaderFromBuf(lg.Raw[i-size : i])
 		tr.DecodeBinary(r)
 		if r.Err != nil {
-			return r.Err
+			return false, r.Err
-		} else if err := f(); err != nil {
+		}
-			return nil
+		cont, err := f()
+		if err != nil {
+			return false, err
+		}
+		if !cont {
+			return false, nil
 		}
 	}
-	return nil
+	return true, nil
 }
 
 // Size returns an amount of transfer written in log provided size of a single transfer.

pkg/core/state/nep5_test.go

@@ -30,13 +30,13 @@ func TestNEP5TransferLog_Append(t *testing.T) {
 
 	i := len(expected) - 1
 	tr := new(NEP5Transfer)
-	err := lg.ForEach(NEP5TransferSize, tr, func() error {
+	cont, err := lg.ForEach(NEP5TransferSize, tr, func() (bool, error) {
 		require.Equal(t, expected[i], tr)
 		i--
-		return nil
+		return true, nil
 	})
 	require.NoError(t, err)
-
+	require.True(t, cont)
 }
 
 func TestNEP5Tracker_EncodeBinary(t *testing.T) {

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) ForEachNEP5Transfer(util.Uint160, *state.NEP5Transfer, func() error) error {
+func (chain testChain) ForEachNEP5Transfer(util.Uint160, *state.NEP5Transfer, func() (bool, error)) error {
 	panic("TODO")
 }
 func (chain testChain) GetNEP5Balances(util.Uint160) *state.NEP5Balances {
@@ -108,7 +108,7 @@ 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 {
+func (chain testChain) ForEachTransfer(util.Uint160, *state.Transfer, func() (bool, error)) error {
 	panic("TODO")
 }
 func (chain testChain) GetScriptHashesForVerifying(*transaction.Transaction) ([]util.Uint160, error) {

pkg/rpc/server/server.go

@@ -544,7 +544,10 @@ func (s *Server) getUTXOTransfers(ps request.Params) (interface{}, *response.Err
 		return nil, response.NewInvalidParamsError("", err)
 	}
 	tr := new(state.Transfer)
-	err = s.chain.ForEachTransfer(addr, tr, func() error {
+	err = s.chain.ForEachTransfer(addr, tr, func() (bool, error) {
+		if tr.Timestamp > end {
+			return true, nil
+		}
 		var count int
 		for _, res := range sent {
 			count += len(res.Transactions)
@@ -552,9 +555,9 @@ func (s *Server) getUTXOTransfers(ps request.Params) (interface{}, *response.Err
 		for _, res := range recv {
 			count += len(res.Transactions)
 		}
-		if tr.Timestamp < start || end != 0 && tr.Timestamp > end ||
+		if tr.Timestamp < start ||
 			(limit != 0 && count >= limit) {
-			return nil
+			return false, nil
 		}
 		assetID := core.GoverningTokenID()
 		if !tr.IsGoverning {
@@ -574,7 +577,7 @@ func (s *Server) getUTXOTransfers(ps request.Params) (interface{}, *response.Err
 			})
 			a.TotalAmount += tr.Amount
 		}
-		return nil
+		return true, nil
 	})
 	if err != nil {
 		return nil, response.NewInternalServerError("", err)
@@ -757,10 +760,13 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err
 		Sent:     []result.NEP5Transfer{},
 	}
 	tr := new(state.NEP5Transfer)
-	err = s.chain.ForEachNEP5Transfer(u, tr, func() error {
+	err = s.chain.ForEachNEP5Transfer(u, tr, func() (bool, error) {
-		if tr.Timestamp < start || tr.Timestamp > end ||
+		if tr.Timestamp > end {
+			return true, nil
+		}
+		if tr.Timestamp < start ||
 			(limit != 0 && (len(bs.Received)+len(bs.Sent) >= limit)) {
-			return nil
+			return false, nil
 		}
 		transfer := result.NEP5Transfer{
 			Timestamp: tr.Timestamp,
@@ -776,7 +782,7 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err
 				transfer.Address = address.Uint160ToString(tr.From)
 			}
 			bs.Received = append(bs.Received, transfer)
-			return nil
+			return true, nil
 		}
 
 		transfer.Amount = strconv.FormatInt(-tr.Amount, 10)
@@ -784,7 +790,7 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err
 			transfer.Address = address.Uint160ToString(tr.To)
 		}
 		bs.Sent = append(bs.Sent, transfer)
-		return nil
+		return true, nil
 	})
 	if err != nil {
 		return nil, response.NewInternalServerError("invalid NEP5 transfer log", err)