stateDB txlog store enhancement

CHANGELOG.md

@@ -54,6 +54,10 @@ Ref: https://keepachangelog.com/en/1.0.0/
 * (encoding) [tharsis#478](https://github.com/tharsis/ethermint/pull/478) Register `Evidence` to amino codec.
 * (rpc) [tharsis#478](https://github.com/tharsis/ethermint/pull/481) Getting the node configuration when calling the `miner` rpc methods.
 
+### Improvements
+
+* (evm) [tharsis#461](https://github.com/tharsis/ethermint/pull/461) Increase performance of `StateDB` transaction log storage (r/w).
+
 ## [v0.5.0] - 2021-08-20
 
 ### State Machine Breaking

x/evm/keeper/grpc_query.go

@@ -232,15 +232,21 @@ func (k Keeper) BlockLogs(c context.Context, req *types.QueryBlockLogsRequest) (
 	ctx := sdk.UnwrapSDKContext(c)
 
 	store := prefix.NewStore(ctx.KVStore(k.storeKey), types.KeyPrefixLogs)
-	txLogs := []types.TransactionLogs{}
+
+	mapOrder := []string{}
+	logs := make(map[string][]*types.Log)
 
 	pageRes, err := query.FilteredPaginate(store, req.Pagination, func(_, value []byte, accumulate bool) (bool, error) {
-		var txLog types.TransactionLogs
+		var txLog types.Log
 		k.cdc.MustUnmarshal(value, &txLog)
 
-		if len(txLog.Logs) > 0 && txLog.Logs[0].BlockHash == req.Hash {
+		if txLog.BlockHash == req.Hash {
 			if accumulate {
-				txLogs = append(txLogs, txLog)
+				if len(logs[txLog.TxHash]) == 0 {
+					mapOrder = append(mapOrder, txLog.TxHash)
+				}
+
+				logs[txLog.TxHash] = append(logs[txLog.TxHash], &txLog)
 			}
 			return true, nil
 		}
@@ -252,8 +258,15 @@ func (k Keeper) BlockLogs(c context.Context, req *types.QueryBlockLogsRequest) (
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	var txsLogs = []types.TransactionLogs{}
+	for _, txHash := range mapOrder {
+		if len(logs[txHash]) > 0 {
+			txsLogs = append(txsLogs, types.TransactionLogs{Hash: txHash, Logs: logs[txHash]})
+		}
+	}
+
 	return &types.QueryBlockLogsResponse{
-		TxLogs:     txLogs,
+		TxLogs:     txsLogs,
 		Pagination: pageRes,
 	}, nil
 }

x/evm/keeper/grpc_query_test.go

@@ -474,7 +474,7 @@ func (suite *KeeperTestSuite) TestQueryBlockLogs() {
 								TxHash:      ethcmn.BytesToHash([]byte("tx_hash_1")).String(),
 								TxIndex:     1,
 								BlockHash:   ethcmn.BytesToHash([]byte("block_hash")).String(),
-								Index:       0,
+								Index:       1,
 								Removed:     false,
 							},
 							{
@@ -485,7 +485,7 @@ func (suite *KeeperTestSuite) TestQueryBlockLogs() {
 								TxHash:      ethcmn.BytesToHash([]byte("tx_hash_1")).String(),
 								TxIndex:     1,
 								BlockHash:   ethcmn.BytesToHash([]byte("block_hash")).String(),
-								Index:       0,
+								Index:       2,
 								Removed:     false,
 							},
 						},

x/evm/keeper/keeper.go

@@ -229,16 +229,30 @@ func (k Keeper) ResetRefundTransient(ctx sdk.Context) {
 // GetAllTxLogs return all the transaction logs from the store.
 func (k Keeper) GetAllTxLogs(ctx sdk.Context) []types.TransactionLogs {
 	store := ctx.KVStore(k.storeKey)
-	iterator := sdk.KVStorePrefixIterator(store, types.KeyPrefixLogs)
-	defer iterator.Close()
+	iter := sdk.KVStorePrefixIterator(store, types.KeyPrefixLogs)
+	defer iter.Close()
+
+	mapOrder := []string{}
+	var mapLogs = make(map[string][]*types.Log)
+	for ; iter.Valid(); iter.Next() {
+		var txLog types.Log
+		k.cdc.MustUnmarshal(iter.Value(), &txLog)
+
+		txlogs := mapLogs[txLog.TxHash]
+		if len(txlogs) == 0 {
+			mapOrder = append(mapOrder, txLog.TxHash)
+		}
 
-	txsLogs := []types.TransactionLogs{}
-	for ; iterator.Valid(); iterator.Next() {
-		var txLog types.TransactionLogs
-		k.cdc.MustUnmarshal(iterator.Value(), &txLog)
+		txlogs = append(txlogs, &txLog)
+		mapLogs[txLog.TxHash] = txlogs
+	}
 
-		// add a new entry
-		txsLogs = append(txsLogs, txLog)
+	txsLogs := []types.TransactionLogs{}
+	for _, txHash := range mapOrder {
+		if len(mapLogs[txHash]) > 0 {
+			txLogs := types.TransactionLogs{Hash: txHash, Logs: mapLogs[txHash]}
+			txsLogs = append(txsLogs, txLogs)
+		}
 	}
 	return txsLogs
 }
@@ -248,31 +262,64 @@ func (k Keeper) GetAllTxLogs(ctx sdk.Context) []types.TransactionLogs {
 func (k Keeper) GetTxLogs(txHash common.Hash) []*ethtypes.Log {
 	store := prefix.NewStore(k.Ctx().KVStore(k.storeKey), types.KeyPrefixLogs)
 
-	bz := store.Get(txHash.Bytes())
-	if len(bz) == 0 {
-		return []*ethtypes.Log{}
-	}
+	// We store the logs with key equal to txHash.Bytes() | sdk.Uint64ToBigEndian(uint64(log.Index)),
+	// therefore, we set the end boundary(excluded) to txHash.Bytes() | uint64.Max -> []byte
+	var end = txHash.Bytes()
+	end = append(end, []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}...)
 
-	var logs types.TransactionLogs
-	k.cdc.MustUnmarshal(bz, &logs)
+	iter := store.Iterator(txHash.Bytes(), end)
+	defer iter.Close()
+
+	logs := []*ethtypes.Log{}
+	for ; iter.Valid(); iter.Next() {
+		var log types.Log
+		k.cdc.MustUnmarshal(iter.Value(), &log)
+		logs = append(logs, log.ToEthereum())
+	}
 
-	return logs.EthLogs()
+	return logs
 }
 
 // SetLogs sets the logs for a transaction in the KVStore.
 func (k Keeper) SetLogs(txHash common.Hash, logs []*ethtypes.Log) {
 	store := prefix.NewStore(k.Ctx().KVStore(k.storeKey), types.KeyPrefixLogs)
 
-	txLogs := types.NewTransactionLogsFromEth(txHash, logs)
-	bz := k.cdc.MustMarshal(&txLogs)
+	for _, log := range logs {
+		var key = txHash.Bytes()
+		key = append(key, sdk.Uint64ToBigEndian(uint64(log.Index))...)
+		txIndexLog := types.NewLogFromEth(log)
+		bz := k.cdc.MustMarshal(txIndexLog)
+		store.Set(key, bz)
+	}
+}
+
+// SetLog sets the log for a transaction in the KVStore.
+func (k Keeper) SetLog(log *ethtypes.Log) {
+	store := prefix.NewStore(k.Ctx().KVStore(k.storeKey), types.KeyPrefixLogs)
+
+	var key = log.TxHash.Bytes()
+	key = append(key, sdk.Uint64ToBigEndian(uint64(log.Index))...)
 
-	store.Set(txHash.Bytes(), bz)
+	txIndexLog := types.NewLogFromEth(log)
+	bz := k.cdc.MustMarshal(txIndexLog)
+	store.Set(key, bz)
 }
 
 // DeleteLogs removes the logs from the KVStore. It is used during journal.Revert.
 func (k Keeper) DeleteTxLogs(ctx sdk.Context, txHash common.Hash) {
 	store := prefix.NewStore(ctx.KVStore(k.storeKey), types.KeyPrefixLogs)
-	store.Delete(txHash.Bytes())
+
+	// We store the logs with key equal to txHash.Bytes() | sdk.Uint64ToBigEndian(uint64(log.Index)),
+	// therefore, we set the end boundary(excluded) to txHash.Bytes() | uint64.Max -> []byte
+	var end = txHash.Bytes()
+	end = append(end, []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}...)
+
+	iter := store.Iterator(txHash.Bytes(), end)
+	defer iter.Close()
+
+	for ; iter.Valid(); iter.Next() {
+		store.Delete(iter.Key())
+	}
 }
 
 // GetLogSizeTransient returns EVM log index on the current block.

x/evm/keeper/statedb.go

@@ -607,11 +607,7 @@ func (k *Keeper) AddLog(log *ethtypes.Log) {
 
 	log.Index = uint(k.GetLogSizeTransient())
 	k.IncreaseLogSizeTransient()
-
-	logs := k.GetTxLogs(log.TxHash)
-	logs = append(logs, log)
-
-	k.SetLogs(log.TxHash, logs)
+	k.SetLog(log)
 
 	k.Logger(k.Ctx()).Debug(
 		"log added",