Better performance of tx deserialization (#229)

2023-06-12 18:21:08 +04:00
parent 83a3ee55a8
commit f878579c7e
11 changed files with 152 additions and 25 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt
@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.data

 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.etherjar.rpc.json.TransactionJson

 class TxContainer(
@@ -48,6 +49,16 @@ class TxContainer(
                tx
            )
        }
+
+        fun from(tx: TransactionJsonSnapshot, raw: ByteArray): TxContainer {
+            return TxContainer(
+                tx.blockNumber,
+                TxId.from(tx.hash),
+                tx.blockHash?.let { BlockId.from(it) },
+                raw,
+                tx
+            )
+        }
    }

    override fun equals(other: Any?): Boolean {
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt
@@ -26,10 +26,10 @@ import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.etherjar.domain.TransactionId
 import io.emeraldpay.etherjar.rpc.RpcException
-import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Qualifier
@@ -159,7 +159,7 @@ class TrackEthereumTx(
            .txByHash().read(tx.txid)
            .onErrorResume(RpcException::class.java) { t ->
                log.warn("Upstream error, ignoring. {}", t.rpcMessage)
-                Mono.empty<TransactionJson>()
+                Mono.empty()
            }
            .flatMap { updateFromBlock(upstream, tx, it) }
            .doOnError { t ->
@@ -212,7 +212,7 @@ class TrackEthereumTx(
            }
    }

-    fun updateFromBlock(upstream: EthereumPosMultiStream, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
+    fun updateFromBlock(upstream: EthereumPosMultiStream, tx: TxDetails, blockTx: TransactionJsonSnapshot): Mono<TxDetails> {
        return if (blockTx.blockNumber != null && blockTx.blockHash != null && blockTx.blockHash != ZERO_BLOCK) {
            val updated = tx.withStatus(
                blockHash = blockTx.blockHash,
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt
@@ -40,11 +40,11 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.etherjar.domain.TransactionId
 import io.emeraldpay.etherjar.domain.Wei
-import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.apache.commons.collections4.Factory
 import org.slf4j.LoggerFactory
@@ -80,8 +80,8 @@ open class EthereumCachingReader(
        }
    }

-    val extractTx = Function<TxContainer, TransactionJson> { tx ->
-        tx.getParsed(TransactionJson::class.java) ?: objectMapper.readValue(tx.json, TransactionJson::class.java)
+    val extractTx = Function<TxContainer, TransactionJsonSnapshot> { tx ->
+        tx.getParsed(TransactionJsonSnapshot::class.java) ?: objectMapper.readValue(tx.json, TransactionJsonSnapshot::class.java)
    }

    val asRaw = Function<SourceContainer, ByteArray> { tx ->
@@ -141,7 +141,7 @@ open class EthereumCachingReader(
        )
    }

-    open fun txByHash(): Reader<TransactionId, TransactionJson> {
+    open fun txByHash(): Reader<TransactionId, TransactionJsonSnapshot> {
        return TransformingReader(
            CompoundReader(
                RekeyingReader(txHashToId, caches.getTxByHash()),
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt
@@ -16,6 +16,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.domain.BlockHash
@@ -24,7 +25,6 @@ import io.emeraldpay.etherjar.domain.Wei
 import io.emeraldpay.etherjar.hex.HexQuantity
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
-import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.apache.commons.collections4.Factory
@@ -79,7 +79,7 @@ class EthereumDirectReader(
                return readWithQuorum(request) // retries were removed because we use NotNullQuorum which handle errors too
                    .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
                    .flatMap { txbytes ->
-                        val tx = objectMapper.readValue(txbytes, TransactionJson::class.java)
+                        val tx = objectMapper.readValue(txbytes, TransactionJsonSnapshot::class.java)
                        if (tx == null) {
                            Mono.empty()
                        } else {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt
@@ -19,8 +19,8 @@ import io.emeraldpay.dshackle.upstream.AbstractChainFees
 import io.emeraldpay.dshackle.upstream.ChainFees
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.etherjar.domain.Wei
-import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
@@ -32,7 +32,7 @@ abstract class EthereumFees(
    upstreams: Multistream,
    private val reader: EthereumCachingReader,
    heightLimit: Int,
-) : AbstractChainFees<EthereumFees.EthereumFee, BlockJson<TransactionRefJson>, TransactionRefJson, TransactionJson>(heightLimit, upstreams, extractTx), ChainFees {
+) : AbstractChainFees<EthereumFees.EthereumFee, BlockJson<TransactionRefJson>, TransactionRefJson, TransactionJsonSnapshot>(heightLimit, upstreams, extractTx), ChainFees {

    companion object {
        private val log = LoggerFactory.getLogger(EthereumFees::class.java)
@@ -42,7 +42,7 @@ abstract class EthereumFees(
        }
    }

-    abstract fun extractFee(block: BlockJson<TransactionRefJson>, tx: TransactionJson): EthereumFee
+    abstract fun extractFee(block: BlockJson<TransactionRefJson>, tx: TransactionJsonSnapshot): EthereumFee

    override fun readFeesAt(height: Long, selector: TxAt<BlockJson<TransactionRefJson>, TransactionRefJson>): Mono<EthereumFee> {
        return reader.blocksByHeightParsed().read(height)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFees.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFees.kt
@@ -17,8 +17,8 @@ package io.emeraldpay.dshackle.upstream.ethereum

 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.etherjar.domain.Wei
-import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import java.util.function.Function
@@ -39,7 +39,7 @@ class EthereumLegacyFees(upstreams: EthereumMultistream, reader: EthereumCaching
            .build()
    }

-    override fun extractFee(block: BlockJson<TransactionRefJson>, tx: TransactionJson): EthereumFee {
+    override fun extractFee(block: BlockJson<TransactionRefJson>, tx: TransactionJsonSnapshot): EthereumFee {
        return EthereumFee(tx.gasPrice, tx.gasPrice, tx.gasPrice, Wei.ZERO)
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt
@@ -18,8 +18,8 @@ package io.emeraldpay.dshackle.upstream.ethereum
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.etherjar.domain.Wei
-import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import java.util.function.Function
@@ -43,7 +43,7 @@ class EthereumPriorityFees(upstreams: Multistream, reader: EthereumCachingReader
                .build()
        }

-    override fun extractFee(block: BlockJson<TransactionRefJson>, tx: TransactionJson): EthereumFee {
+    override fun extractFee(block: BlockJson<TransactionRefJson>, tx: TransactionJsonSnapshot): EthereumFee {
        val baseFee = block.baseFeePerGas ?: Wei.ZERO
        if (tx.type == 2) {
            // an EIP-1559 Transaction provides Max and Priority fee