StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

changes from upstream: removed RpcReader and use better flow for tx receipt

Browse Source
This commit is contained in:
Termina1
2022-12-09 23:04:50 +02:00
parent c41d8c85ba
commit a4f0d44105
19 changed files with 139 additions and 372 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt vendored
View File

@@ -92,7 +92,7 @@ open class Caches(
open fun cacheReceipt(tag: Tag, data: DefaultContainer<TransactionReceiptJson>) {
val currentHeight = head?.getCurrentHeight()
if (currentHeight != null && data.height != null && memReceipts.acceptsRecentBlocks(currentHeight - data.height)) {
memReceipts.add(data)
memReceipts.add(data).subscribe()
}
// TODO move subscription to the caller
redisReceipts?.add(data)?.subscribe()

8
src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt
View File

@@ -18,11 +18,11 @@ package io.emeraldpay.dshackle.data
import org.slf4j.LoggerFactory
class DefaultContainer<T>(
val txId: TxId?,
val blockId: BlockId?,
val height: Long?,
val txId: TxId? = null,
val blockId: BlockId? = null,
val height: Long? = null,
json: ByteArray,
parsed: T
parsed: T? = null
) : SourceContainer(json, parsed) {
companion object {

54
src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt
View File

@@ -1,54 +0,0 @@
/**
* Copyright (c) 2021 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.reader
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
/**
* Reader that requests data through upstream RPC using provided JSON RPC request builder
*/
class RpcReader<T>(
private val up: Multistream,
private val paramsBuilder: (T) -> JsonRpcRequest
) : Reader<T, ByteArray> {
companion object {
private val log = LoggerFactory.getLogger(RpcReader::class.java)
/**
* Common reader that just passes key as a parameter with the specified method. The key must be serializable to JSON.
* @param method RPC method to use
*/
fun <T> basicRequest(up: Multistream, method: String): RpcReader<T> {
return RpcReader(up) { key ->
JsonRpcRequest(method, listOf(key))
}
}
}
override fun read(key: T): Mono<ByteArray> {
return up.getDirectApi(Selector.empty)
.flatMap { rdr ->
rdr.read(paramsBuilder(key)).flatMap {
it.requireResult()
}
}
}
}

12
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt
View File

@@ -28,7 +28,6 @@ import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.CompoundReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.RekeyingReader
import io.emeraldpay.dshackle.reader.RpcReader
import io.emeraldpay.dshackle.reader.TransformingReader
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.Multistream
@@ -45,16 +44,16 @@ import org.slf4j.LoggerFactory
import java.util.function.Function
/**
* Reader for the common operations, that wraps caches + native call with quorum verification
* Reader for the common operations, that use the cache when data is available or a native call with quorum verification
*/
open class EthereumReader(
open class EthereumCachingReader(
private val up: Multistream,
private val caches: Caches,
private val callMethodsFactory: Factory<CallMethods>
) : Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(EthereumReader::class.java)
private val log = LoggerFactory.getLogger(EthereumCachingReader::class.java)
}
private val objectMapper: ObjectMapper = Global.objectMapper
@@ -157,10 +156,9 @@ open class EthereumReader(
}
fun receipts(): Reader<TxId, ByteArray> {
// TODO put into cache
val requested = RekeyingReader(
{ txid: TxId -> txid.toHexWithPrefix() },
RpcReader.basicRequest(up, "eth_getTransactionReceipt")
{ txid: TxId -> TransactionId.from(txid.value) },
directReader.receiptReader
)
return CompoundReader(
caches.getReceipts(),

31
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt
View File

@@ -6,7 +6,10 @@ import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CurrentBlockCache
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.DefaultContainer
import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Multistream
@@ -22,6 +25,7 @@ import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.emeraldpay.etherjar.rpc.json.BlockJson
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
import org.slf4j.LoggerFactory
@@ -51,6 +55,7 @@ class EthereumDirectReader(
val blockByHeightReader: Reader<Long, BlockContainer>
val txReader: Reader<TransactionId, TxContainer>
val balanceReader: Reader<Address, Wei>
val receiptReader: Reader<TransactionId, ByteArray>
init {
blockReader = object : Reader<BlockHash, BlockContainer> {
@@ -107,6 +112,32 @@ class EthereumDirectReader(
}
}
}
receiptReader = object : Reader<TransactionId, ByteArray> {
override fun read(key: TransactionId): Mono<ByteArray> {
val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex()))
return readWithQuorum(request)
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Receipt not read $key")))
.doOnNext { json ->
try {
// Caching needs some additional data (ex. Height) to make a decision on how long and where to cache
// So we have to parse the JSON here and extract reference data
val receipt = objectMapper.readValue(json, TransactionReceiptJson::class.java)
caches.cacheReceipt(
Caches.Tag.REQUESTED,
DefaultContainer(
txId = TxId.from(key),
blockId = BlockId.from(receipt.blockHash),
height = receipt.blockNumber,
json = json,
parsed = receipt
)
)
} catch (t: Throwable) {
log.warn("Failed to cache Tx Receipt", t)
}
}
}
}
}
@Suppress("UNCHECKED_CAST")

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt
View File

@@ -30,7 +30,7 @@ import java.util.function.Function
abstract class EthereumFees(
upstreams: Multistream,
private val reader: EthereumReader,
private val reader: EthereumCachingReader,
heightLimit: Int,
) : AbstractChainFees<EthereumFees.EthereumFee, BlockJson<TransactionRefJson>, TransactionRefJson, TransactionJson>(heightLimit, upstreams, extractTx), ChainFees {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFees.kt
View File

@@ -23,7 +23,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import java.util.function.Function
class EthereumLegacyFees(upstreams: EthereumMultistream, reader: EthereumReader, heightLimit: Int) :
class EthereumLegacyFees(upstreams: EthereumMultistream, reader: EthereumCachingReader, heightLimit: Int) :
EthereumFees(upstreams, reader, heightLimit) {
companion object {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
View File

@@ -7,7 +7,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
import reactor.core.publisher.Flux
interface EthereumLikeMultistream : Upstream {
fun getReader(): EthereumReader
fun getReader(): EthereumCachingReader
fun getSubscribe(): EthereumSubscribe
fun getHead(mather: Selector.Matcher): Head

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
View File

@@ -47,8 +47,9 @@ open class EthereumMultistream(
private val filteredHeads: MutableMap<String, Head> =
ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory())
private val subscribe = EthereumSubscribe(this)
private val supportsEIP1559 = when (chain) {
Chain.ETHEREUM, Chain.TESTNET_ROPSTEN, Chain.TESTNET_GOERLI, Chain.TESTNET_RINKEBY -> true
else -> false
@@ -82,7 +83,7 @@ open class EthereumMultistream(
return super.isRunning() || reader.isRunning()
}
override fun getReader(): EthereumReader {
override fun getReader(): EthereumCachingReader {
return reader
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt
View File

@@ -24,7 +24,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import java.util.function.Function
class EthereumPriorityFees(upstreams: Multistream, reader: EthereumReader, heightLimit: Int) :
class EthereumPriorityFees(upstreams: Multistream, reader: EthereumCachingReader, heightLimit: Int) :
EthereumFees(upstreams, reader, heightLimit) {
companion object {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
View File

@@ -34,10 +34,10 @@ import java.math.BigInteger
* It provides data only if it's available through the router (cached, head, etc).
* If data is not available locally then it returns `empty`; at this case the caller should call the remote node for actual data.
*
* @see EthereumReader
* @see EthereumCachingReader
*/
class LocalCallRouter(
private val reader: EthereumReader,
private val reader: EthereumCachingReader,
private val methods: CallMethods,
private val head: Head
) : Reader<JsonRpcRequest, JsonRpcResponse> {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
View File

@@ -45,7 +45,7 @@ open class EthereumPosMultiStream(
private var head: Head? = null
private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory())
private val feeEstimation = EthereumPriorityFees(this, reader, 256)
private val subscribe = EthereumSubscribe(this)
private val filteredHeads: MutableMap<String, Head> =
@@ -77,7 +77,7 @@ open class EthereumPosMultiStream(
return super.isRunning() || reader.isRunning()
}
override fun getReader(): EthereumReader {
override fun getReader(): EthereumCachingReader {
return reader
}