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

Merge pull request #77 from p2p-org/upstream-get-transaction-receipt

Browse Source 
changes from upstream: removed RpcReader and use better flow for tx r…
This commit is contained in:
Vyacheslav Shebanov
2022-12-10 16:03:49 +02:00
committed by GitHub
parent c41d8c85ba 8c34a5ce2f
commit af69db7594
26 changed files with 153 additions and 661 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()
}
}
}
}

18
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -52,7 +52,6 @@ abstract class Multistream(
val chain: Chain,
private val upstreams: MutableList<Upstream>,
val caches: Caches,
val postprocessor: RequestPostprocessor
) : Upstream, Lifecycle {
companion object {
@@ -144,23 +143,6 @@ abstract class Multistream(
return FilteredApis(chain, upstreams, matcher, i)
}
/**
* Finds an API that executed directly on a remote.
*/
open fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
val apis = getApiSource(matcher)
apis.request(1)
return Mono.from(apis)
.map(Upstream::getApi)
.map {
RequestPostprocessor.wrap(
it,
postprocessor
)
} // TODO do it on upstream init, not each time it's called
.switchIfEmpty(Mono.error(Exception("No API available for $chain")))
}
abstract fun getFeeEstimation(): ChainFees
/**

40
src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt
View File

@@ -1,40 +0,0 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import reactor.core.publisher.Mono
interface RequestPostprocessor {
fun onReceive(method: String, params: List<Any?>, json: ByteArray)
class Empty : RequestPostprocessor {
override fun onReceive(method: String, params: List<Any?>, json: ByteArray) {}
}
companion object {
fun wrap(
reader: Reader<JsonRpcRequest, JsonRpcResponse>,
processor: RequestPostprocessor
): Reader<JsonRpcRequest, JsonRpcResponse> {
return Wrapper(reader, processor)
}
}
class Wrapper(
private val reader: Reader<JsonRpcRequest, JsonRpcResponse>,
private val processor: RequestPostprocessor
) : Reader<JsonRpcRequest, JsonRpcResponse> {
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
return reader.read(key)
.doOnNext {
if (it.hasResult()) {
val result = it.getResult()
processor.onReceive(key.method, key.params, result)
}
}
}
}
}

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
View File

@@ -32,8 +32,8 @@ import reactor.core.publisher.Mono
open class BitcoinMultistream(
chain: Chain,
private val sourceUpstreams: MutableList<BitcoinUpstream>,
caches: Caches
) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty()), Lifecycle {
caches: Caches,
) : Multistream(chain, sourceUpstreams as MutableList<Upstream>, caches), Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java)
@@ -95,6 +95,16 @@ open class BitcoinMultistream(
onHeadUpdated(head)
return head
}
/**
* Finds an API that executed directly on a remote.
*/
open fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
val apis = getApiSource(matcher)
apis.request(1)
return Mono.from(apis)
.map(Upstream::getApi)
.switchIfEmpty(Mono.error(Exception("No API available for $chain")))
}
override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return Mono.just(callRouter)

60
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt
View File

@@ -1,60 +0,0 @@
/**
* Copyright (c) 2020 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.upstream.ethereum
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.DefaultContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.upstream.RequestPostprocessor
import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
import org.slf4j.LoggerFactory
class CacheRequested(
private val caches: Caches
) : RequestPostprocessor {
companion object {
private val log = LoggerFactory.getLogger(CacheRequested::class.java)
}
override fun onReceive(method: String, params: List<Any?>, json: ByteArray) {
try {
if (method == "eth_getTransactionReceipt") {
cacheTxReceipt(params, json)
}
} catch (e: Throwable) {
log.warn("Failed to cache result", e)
}
}
fun cacheTxReceipt(params: List<Any?>, json: ByteArray) {
if (params.size != 1) {
return
}
// note: json could be a `null` value
val parsed = Global.objectMapper.readValue(json, TransactionReceiptJson::class.java) ?: return
val value = DefaultContainer<TransactionReceiptJson>(
TxId.from(parsed.transactionHash),
BlockId.from(parsed.blockHash),
parsed.blockNumber,
json,
parsed
)
caches.cacheReceipt(Caches.Tag.REQUESTED, value)
}
}

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

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

@@ -37,7 +37,7 @@ open class EthereumMultistream(
chain: Chain,
val upstreams: MutableList<EthereumUpstream>,
caches: Caches
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)), EthereumLikeMultistream {
) : Multistream(chain, upstreams as MutableList<Upstream>, caches), EthereumLikeMultistream {
companion object {
private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)
@@ -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> {

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

@@ -37,7 +37,7 @@ open class EthereumPosMultiStream(
chain: Chain,
val upstreams: MutableList<EthereumPosUpstream>,
caches: Caches
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)), EthereumLikeMultistream {
) : Multistream(chain, upstreams as MutableList<Upstream>, caches), EthereumLikeMultistream {
companion object {
private val log = LoggerFactory.getLogger(EthereumPosMultiStream::class.java)
@@ -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
}

2
src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy vendored
View File

@@ -262,6 +262,6 @@ class CachesSpec extends Specification {
then:
1 * head.getCurrentHeight() >> 0xccf6e2
1 * receiptMemCache.acceptsRecentBlocks(0) >> true
1 * receiptMemCache.add(receiptContainer)
1 * receiptMemCache.add(receiptContainer) >> Mono.empty().then()
}
}

7
src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy
View File

@@ -29,9 +29,10 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumReader
import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Chain
import org.jetbrains.annotations.NotNull
class MultistreamHolderMock implements MultistreamHolder {
@@ -87,7 +88,7 @@ class MultistreamHolderMock implements MultistreamHolder {
static class EthereumMultistreamMock extends EthereumPosMultiStream {
EthereumReader customReader = null
EthereumCachingReader customReader = null
CallMethods customMethods = null
Head customHead = null
@@ -104,7 +105,7 @@ class MultistreamHolderMock implements MultistreamHolder {
}
@Override
EthereumReader getReader() {
EthereumCachingReader getReader() {
if (customReader != null) {
return customReader
}

62
src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
View File

@@ -180,26 +180,6 @@ class MultistreamSpec extends Specification {
!act
}
def "Call postprocess after api use"() {
setup:
def request = new JsonRpcRequest("test_foo", [1], 1, null)
def api = TestingCommons.api()
api.answer("test_foo", [1], "test")
def postprocessor = Mock(RequestPostprocessor)
def up = TestingCommons.upstream(api)
def multistream = new TestMultistream([up], postprocessor)
when:
def rdr = multistream.getDirectApi(Selector.empty).block(Duration.ofSeconds(1))
def act = rdr.read(request).block(Duration.ofSeconds(1))
then:
act != null
act.hasResult()
act.resultAsProcessedString == "test"
1 * postprocessor.onReceive("test_foo", [1], "\"test\"".bytes)
}
def "Filter upstream matching selector single"() {
setup:
@@ -341,48 +321,6 @@ class MultistreamSpec extends Specification {
.expectComplete()
.verify(Duration.ofSeconds(1))
}
class TestMultistream extends Multistream {
TestMultistream(List<Upstream> upstreams, @NotNull RequestPostprocessor postprocessor) {
super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor)
}
@Override
Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getRoutedApi(@NotNull Selector.Matcher matcher) {
return null
}
@Override
Head updateHead() {
return null
}
@Override
void setHead(@NotNull Head head) {
}
@Override
Head getHead() {
return null
}
@Override
Collection<UpstreamsConfig.Labels> getLabels() {
return null
}
public <T extends Upstream> T cast(Class<T> selfType) {
return this
}
@Override
ChainFees getFeeEstimation() {
return null
}
}
class TestEthereumPosMultistream extends EthereumPosMultiStream {
TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumPosUpstream> upstreams, @NotNull Caches caches) {

47
src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy
View File

@@ -1,47 +0,0 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import reactor.core.publisher.Mono
import spock.lang.Specification
import java.time.Duration
class RequestPostprocessorSpec extends Specification {
def "Wrappers calls onReceive for a value"() {
setup:
def request = new JsonRpcRequest("test_foo", [1], 1, null)
def processor = Mock(RequestPostprocessor)
def api = TestingCommons.api()
api.answer("test_foo", [1], "test")
def wrapped = new RequestPostprocessor.Wrapper(api, processor)
when:
def act = wrapped.read(request).block(Duration.ofSeconds(1))
then:
act.hasResult()
act.resultAsProcessedString == "test"
1 * processor.onReceive("test_foo", [1], "\"test\"".bytes)
}
def "Wrappers doesn't call onReceive for no value"() {
setup:
def request = new JsonRpcRequest("test_foo", [1], 1, null)
def processor = Mock(RequestPostprocessor)
Reader<JsonRpcRequest, JsonRpcResponse> reader = Mock(Reader) {
1 * it.read(request) >> Mono.empty()
}
def wrapped = new RequestPostprocessor.Wrapper(reader, processor)
when:
def act = wrapped.read(request).block(Duration.ofSeconds(1))
then:
act == null
0 * processor.onReceive(_, _, _)
}
}

58
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy
View File

@@ -1,58 +0,0 @@
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.data.DefaultContainer
import spock.lang.Specification
class CacheRequestedSpec extends Specification {
def "Do nothing if unsupported method"() {
setup:
def caches = Mock(Caches)
CacheRequested instance = new CacheRequested(caches)
when:
instance.onReceive("eth_hashrate", [], '"0x38a"'.bytes)
then:
0 * caches._(*_)
}
def "Caches tx receipt"() {
setup:
def caches = Mock(Caches)
CacheRequested instance = new CacheRequested(caches)
def json = '''{
"blockHash": "0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae",
"blockNumber": "0x213b8a",
"contractAddress": null,
"cumulativeGasUsed": "0x5208",
"gasUsed": "0x5208",
"logs": [],
"transactionHash": "0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197",
"transactionIndex": "0x00"
}'''.bytes
when:
instance.onReceive("eth_getTransactionReceipt", ["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json)
then:
1 * caches.cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer it ->
it.height == 0x213b8a &&
it.txId.toHex() == "5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197" &&
it.json == json
})
}
def "Ignore null"() {
setup:
def caches = Mock(Caches)
CacheRequested instance = new CacheRequested(caches)
def json = 'null'.bytes
when:
instance.cacheTxReceipt(["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json)
then:
0 * caches.cacheReceipt(_, _)
}
}

73
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReaderSpec.groovy → src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy
View File

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CurrentBlockCache
import io.emeraldpay.dshackle.data.DefaultContainer
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.quorum.QuorumRpcReader
import io.emeraldpay.dshackle.reader.Reader
@@ -12,6 +13,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
import io.emeraldpay.etherjar.domain.Address
import io.emeraldpay.etherjar.domain.BlockHash
import io.emeraldpay.etherjar.domain.TransactionId
@@ -22,7 +24,6 @@ import org.apache.commons.collections4.Factory
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
import spock.lang.Specification
import java.time.Duration
import java.time.Instant
@@ -171,6 +172,76 @@ class EthereumDirectReaderSpec extends Specification {
.verify(Duration.ofSeconds(1))
}
def "Reads tx receipt"() {
setup:
def json = new TransactionReceiptJson().tap {
transactionHash = TransactionId.from(hash1)
blockNumber = 100
blockHash = BlockHash.from(hash1)
}
def up = Mock(Multistream) {
1 * getApiSource(_) >> Stub(ApiSource)
}
def calls = Mock(Factory) {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
}
EthereumDirectReader reader = new EthereumDirectReader(
up, Caches.default(), new CurrentBlockCache(), calls
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList<Byte>()
)
)
}
}
when:
def act = reader.receiptReader.read(TransactionId.from(hash1))
.block(Duration.ofSeconds(1))
.with { new String(it) }
then:
act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}'
}
def "Puts tx receipt in cache after reading"() {
setup:
def json = new TransactionReceiptJson().tap {
transactionHash = TransactionId.from(hash1)
blockNumber = 100
blockHash = BlockHash.from(hash1)
}
def up = Mock(Multistream) {
1 * getApiSource(_) >> Stub(ApiSource)
}
def calls = Mock(Factory) {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
}
def caches = Mock(Caches) {
// note that the Caches needs a Height value, otherwise it's not cached
1 * cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer data -> data.txId.toHex() == hash1.substring(2) && data.height == 100 })
}
EthereumDirectReader reader = new EthereumDirectReader(
up, caches, new CurrentBlockCache(), calls
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList<Byte>()
)
)
}
}
when:
def act = reader.receiptReader.read(TransactionId.from(hash1))
.block(Duration.ofSeconds(1))
.with { new String(it) }
then:
act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}'
}
def "Produce empty on non-existing tx"() {
setup:
def up = Mock(Multistream) {

2
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFeesSpec.groovy
View File

@@ -30,7 +30,7 @@ class EthereumLegacyFeesSpec extends Specification {
it.gasPrice = Wei.ofUnits(8, Wei.Unit.GWEI)
}
def fees = new EthereumLegacyFees(Stub(EthereumMultistream), Stub(EthereumReader), 10)
def fees = new EthereumLegacyFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10)
when:
def act = fees.extractFee(block, tx)
then:

8
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFeesSpec.groovy
View File

@@ -44,7 +44,7 @@ class EthereumPriorityFeesSpec extends Specification {
it.maxPriorityFeePerGas = Wei.ofUnits(5.0001, Wei.Unit.GWEI)
}
def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumReader), 10)
def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10)
when:
def act = fees.extractFee(block, tx)
then:
@@ -66,7 +66,7 @@ class EthereumPriorityFeesSpec extends Specification {
it.gasPrice = Wei.from("0x198286458f")
}
def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumReader), 10)
def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10)
when:
def act = fees.extractFee(block, tx)
then:
@@ -83,7 +83,7 @@ class EthereumPriorityFeesSpec extends Specification {
new EthereumFees.EthereumFee(Wei.ofEthers(0.75), Wei.ofEthers(0.5), Wei.ofEthers(0.75), Wei.ofEthers(0.5)),
new EthereumFees.EthereumFee(Wei.ofEthers(0.6), Wei.ofEthers(0.2), Wei.ofEthers(0.6), Wei.ofEthers(0.5)),
]
def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumReader), 10)
def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10)
when:
def act = Flux.fromIterable(inputs)
.transform(fees.feeAggregation(ChainFees.Mode.AVG_LAST))
@@ -127,7 +127,7 @@ class EthereumPriorityFeesSpec extends Specification {
1 * getCurrentHeight() >> 13756007
}
}
def reader = Mock(EthereumReader) {
def reader = Mock(EthereumCachingReader) {
_ * it.blocksByHeightParsed() >> Mock(Reader) {
1 * it.read(13756006) >> Mono.just(block1)
1 * it.read(13756007) >> Mono.just(block2)

281
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy
View File

@@ -1,281 +0,0 @@
/**
* Copyright (c) 2020 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.upstream.ethereum
import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.ReceiptRedisCache
import io.emeraldpay.dshackle.cache.TxMemCache
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock
import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.Chain
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.BlockJson
import io.emeraldpay.etherjar.rpc.json.TransactionJson
import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
import org.apache.commons.collections4.Factory
import org.apache.commons.collections4.functors.ConstantFactory
import reactor.core.publisher.Mono
import spock.lang.Specification
import java.time.Instant
class EthereumReaderSpec extends Specification {
def blockId = BlockId.from("f85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")
def blockJson = new BlockJson<TransactionRefJson>().tap { blockJson ->
blockJson.hash = BlockHash.from(blockId.value)
blockJson.totalDifficulty = BigInteger.ONE
blockJson.number = 101
blockJson.timestamp = Instant.ofEpochSecond(100000000)
blockJson.transactions = []
blockJson.uncles = []
}
def txId = BlockId.from("a38e7b4d456777c94b46c61a1e4cf52fbdd92acc4444719d1fad77005698c221")
def txJson = new TransactionJson().tap { json ->
json.hash = TransactionId.from(txId.value)
json.blockHash = blockJson.hash
json.blockNumber = blockJson.number
}
Factory<CallMethods> calls = ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
def "Block by Id reads from cache"() {
setup:
def memCache = Mock(BlocksMemCache) {
1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson))
}
def caches = Caches.newBuilder()
.setBlockByHash(memCache)
.build()
def reader = new EthereumReader(Stub(Multistream), caches, calls)
when:
def act = reader.blocksByIdParsed().read(blockId).block()
then:
act == blockJson
}
def "Block by Id reads from api if cache is empty"() {
setup:
def memCache = Mock(BlocksMemCache) {
1 * read(blockId) >> Mono.empty()
}
def caches = Caches.newBuilder()
.setBlockByHash(memCache)
.build()
def api = TestingCommons.api()
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
def upstream = TestingCommons.multistream(api)
def reader = new EthereumReader(upstream, caches, calls)
when:
def act = reader.blocksByIdParsed().read(blockId).block()
then:
act == blockJson
}
def "Block by Id reads from api if cache failed"() {
setup:
def memCache = Mock(BlocksMemCache) {
1 * read(blockId) >> Mono.error(new IllegalStateException("Test error"))
}
def caches = Caches.newBuilder()
.setBlockByHash(memCache)
.build()
def api = TestingCommons.api()
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
def upstream = TestingCommons.multistream(api)
def reader = new EthereumReader(upstream, caches, calls)
when:
def act = reader.blocksByIdParsed().read(blockId).block()
then:
act == blockJson
}
def "Block by Hash reads from cache"() {
setup:
def memCache = Mock(BlocksMemCache) {
1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson))
}
def caches = Caches.newBuilder()
.setBlockByHash(memCache)
.build()
def reader = new EthereumReader(Stub(Multistream), caches, calls)
when:
def act = reader.blocksByHashParsed().read(blockJson.hash).block()
then:
act == blockJson
}
def "Block by Hash reads from api if cache is empty"() {
setup:
def memCache = Mock(BlocksMemCache) {
1 * read(blockId) >> Mono.empty()
}
def caches = Caches.newBuilder()
.setBlockByHash(memCache)
.build()
def api = TestingCommons.api()
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
def upstream = TestingCommons.multistream(api)
def reader = new EthereumReader(upstream, caches, calls)
when:
def act = reader.blocksByHashParsed().read(blockJson.hash).block()
then:
act == blockJson
}
def "Tx by Hash reads from cache"() {
setup:
def memCache = Mock(TxMemCache) {
1 * read(txId) >> Mono.just(TxContainer.from(txJson))
}
def caches = Caches.newBuilder()
.setTxByHash(memCache)
.build()
def reader = new EthereumReader(Stub(Multistream), caches, calls)
when:
def act = reader.txByHash().read(txJson.hash).block()
then:
act == txJson
}
def "Tx by Hash reads from api if cache is empty"() {
setup:
def memCache = Mock(TxMemCache) {
1 * read(txId) >> Mono.empty()
}
def caches = Caches.newBuilder()
.setTxByHash(memCache)
.build()
def api = TestingCommons.api()
api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson)
def upstream = TestingCommons.multistream(api)
def reader = new EthereumReader(upstream, caches, calls)
when:
def act = reader.txByHash().read(txJson.hash).block()
then:
act == txJson
}
def "Caches balance until block mined"() {
setup:
def api = TestingCommons.api()
// no height
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0x10")
// height 101 + 1 => 102 => 0x66
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "0x66"], "0xff")
EthereumPosRpcUpstreamMock upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api)
def upstreams = TestingCommons.multistream(upstream)
def reader = new EthereumReader(upstreams, Caches.default(), calls)
reader.start()
when:
def act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block()
then:
act == Wei.from("0x10")
when:
//now it should use cached value, without actual request
act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block()
then:
act == Wei.from("0x10")
when:
//move head forward, which should erase cache
def block2 = blockJson.copy().tap {
it.number++
it.totalDifficulty = BigInteger.TWO
}
upstream.nextBlock(BlockContainer.from(block2))
Thread.sleep(50)
act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block()
then:
act == Wei.from("0xff")
}
def "Read receipt from upstream if cache is empty"() {
setup:
def api = TestingCommons.api()
api.answerOnce("eth_getTransactionReceipt", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"], [
transactionHash: "0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"
])
EthereumPosRpcUpstreamMock upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api)
def upstreams = TestingCommons.multistream(upstream)
def reader = new EthereumReader(upstreams, Caches.default(), calls)
reader.start()
when:
def act = reader.receipts().read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")).block()
then:
act != null
new String(act) == '{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}'
}
def "Read receipt from cache if available"() {
setup:
def api = TestingCommons.api()
EthereumPosRpcUpstreamMock upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api)
def upstreams = TestingCommons.multistream(upstream)
def receiptCache = Mock(ReceiptRedisCache) {
1 * it.read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")) >>
Mono.just('{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}'.bytes)
}
def cashes = Caches.newBuilder()
.setReceipts(receiptCache)
.build()
def reader = new EthereumReader(upstreams, cashes, calls)
reader.start()
when:
def act = reader.receipts().read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")).block()
then:
act != null
new String(act) == '{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}'
api.calls.get() == 0
}
}

10
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy
View File

@@ -23,7 +23,7 @@ class LocalCallRouterSpec extends Specification {
setup:
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(
new EthereumReader(
new EthereumCachingReader(
TestingCommons.multistream(TestingCommons.api()),
Caches.default(),
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
@@ -41,7 +41,7 @@ class LocalCallRouterSpec extends Specification {
setup:
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(
new EthereumReader(
new EthereumCachingReader(
TestingCommons.multistream(TestingCommons.api()),
Caches.default(),
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
@@ -61,7 +61,7 @@ class LocalCallRouterSpec extends Specification {
def head = Mock(Head) {
1 * getCurrentHeight() >> 101L
}
def reader = Mock(EthereumReader) {
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont() >> new EmptyReader<>()
_ * txByHashAsCont() >> new EmptyReader<>()
1 * blocksByHeightAsCont() >> Mock(Reader) {
@@ -87,7 +87,7 @@ class LocalCallRouterSpec extends Specification {
def "getBlockByNumber with earliest uses 0 block"() {
setup:
def head = Stub(Head) {}
def reader = Mock(EthereumReader) {
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont() >> new EmptyReader<>()
_ * txByHashAsCont() >> new EmptyReader<>()
1 * blocksByHeightAsCont() >> Mock(Reader) {
@@ -113,7 +113,7 @@ class LocalCallRouterSpec extends Specification {
def "getBlockByNumber fetches the block"() {
setup:
def head = Stub(Head) {}
def reader = Mock(EthereumReader) {
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont() >> new EmptyReader<>()
_ * txByHashAsCont() >> new EmptyReader<>()
1 * blocksByHeightAsCont() >> Mock(Reader) {