diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt index 8aac7b85..179089cc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -33,6 +33,7 @@ open class Caches( private val memBlocksByHash: BlocksMemCache, private val blocksByHeight: HeightCache, private val memTxsByHash: TxMemCache, + private val memReceipts: ReceiptMemCache, private val redisBlocksByHash: BlocksRedisCache?, private val redisTxsByHash: TxRedisCache?, private val redisReceipts: ReceiptRedisCache?, @@ -59,6 +60,8 @@ open class Caches( private val txsByHash: Reader private val receiptByHash: Reader + private var head: Head? = null + init { blocksByHash = if (redisBlocksByHash == null) { memBlocksByHash @@ -70,26 +73,24 @@ open class Caches( } else { CompoundReader(memTxsByHash, redisTxsByHash) } - receiptByHash = redisReceipts ?: EmptyReader() + receiptByHash = if (redisReceipts == null) { + memReceipts + } else { + CompoundReader(memReceipts, redisReceipts) + } } fun setHead(head: Head) { + this.head = head redisTxsByHash?.head = head redisReceipts?.head = head } - /** - * Cache data that was just requested - */ - fun cacheRequested(data: Any) { - if (data is TxContainer) { - cache(Tag.REQUESTED, data) - } else if (data is BlockContainer) { - cache(Tag.REQUESTED, data) - } - } - open fun cacheReceipt(tag: Tag, data: DefaultContainer) { + val currentHeight = head?.getCurrentHeight() + if (currentHeight != null && data.height != null && memReceipts.acceptsRecentBlocks(currentHeight - data.height)) { + memReceipts.add(data) + } //TODO move subscription to the caller redisReceipts?.add(data)?.subscribe() } @@ -165,6 +166,7 @@ open class Caches( memBlocksByHash.get(blockId)?.let { block -> memTxsByHash.evict(block) redisTxsByHash?.evict(block) + memReceipts.evict(block) evicted = true } if (!evicted) { @@ -224,6 +226,7 @@ open class Caches( private var blocksByHash: BlocksMemCache? = null private var blocksByHeight: HeightCache? = null private var txsByHash: TxMemCache? = null + private var receipts: ReceiptMemCache? = null private var redisBlocksByHash: BlocksRedisCache? = null private var redisTxsByHash: TxRedisCache? = null private var redisReceiptCache: ReceiptRedisCache? = null @@ -259,6 +262,11 @@ open class Caches( return this } + fun setReceipts(cache: ReceiptMemCache): Builder { + this.receipts = cache + return this + } + fun setHeightByHash(cache: HeightByHashRedisCache): Builder { redisHeightByHashCache = cache return this @@ -274,7 +282,10 @@ open class Caches( if (txsByHash == null) { txsByHash = TxMemCache() } - return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, + if (receipts == null) { + receipts = ReceiptMemCache() + } + return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, receipts!!, redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptMemCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptMemCache.kt new file mode 100644 index 00000000..1724683f --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptMemCache.kt @@ -0,0 +1,64 @@ +/** + * 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.cache + +import com.github.benmanes.caffeine.cache.Caffeine +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.DefaultContainer +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono + +/** + * Keeps receipts for recent blocks in memory + */ +open class ReceiptMemCache( + // how many blocks to keeps in memory + val blocks: Int = 6 +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(ReceiptMemCache::class.java) + } + + private val mapping = Caffeine.newBuilder() + .maximumSize(blocks * 200L) + .build() + + open fun evict(block: BlockContainer) { + block.transactions.forEach { + mapping.invalidate(it) + } + } + + override fun read(key: TxId): Mono { + return mapping.getIfPresent(key)?.let { Mono.just(it) } ?: Mono.empty() + } + + open fun add(receipt: DefaultContainer): Mono { + if (receipt.txId != null && receipt.json != null) { + mapping.put(receipt.txId, receipt.json) + } + return Mono.empty() + } + + open fun acceptsRecentBlocks(heightDelta: Long): Boolean { + return blocks <= heightDelta && heightDelta >= 0 + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index e79947e8..788ab66a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -200,12 +200,6 @@ open class NativeCall( .map { CallResult(ctx.id, it.value, null) } - .doOnNext { - it.result?.let { value -> - ctx.upstream.postprocessor - .onReceive(ctx.payload.method, ctx.payload.params, value) - } - } .onErrorResume { t -> val failure = if (t is CallFailure) { CallResult.fail(t.id, t.reason) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index f1816dd5..462a2e35 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -140,6 +140,7 @@ abstract class Multistream( 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"))) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt index e8d7947b..5f21123e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt @@ -1,10 +1,38 @@ 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, json: ByteArray) + fun onReceive(method: String, params: List, json: ByteArray) class Empty : RequestPostprocessor { - override fun onReceive(method: String, params: List, json: ByteArray) {} + override fun onReceive(method: String, params: List, json: ByteArray) {} + } + + companion object { + fun wrap(reader: Reader, processor: RequestPostprocessor): Reader { + return Wrapper(reader, processor) + } + } + + class Wrapper( + private val reader: Reader, + private val processor: RequestPostprocessor + ) : Reader { + + override fun read(key: JsonRpcRequest): Mono { + return reader.read(key) + .doOnNext { + if (it.hasResult()) { + val result = it.getResult() + processor.onReceive(key.method, key.params, result) + } + } + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index 8a0fdafc..43ef6b40 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt @@ -26,6 +26,7 @@ class Selector { companion object { + @JvmStatic val empty = EmptyMatcher() @JvmStatic diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt index 7d728f05..4b864916 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt @@ -20,6 +20,7 @@ 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.Head import io.emeraldpay.dshackle.upstream.RequestPostprocessor import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import org.slf4j.LoggerFactory @@ -32,7 +33,7 @@ class CacheRequested( private val log = LoggerFactory.getLogger(CacheRequested::class.java) } - override fun onReceive(method: String, params: List, json: ByteArray) { + override fun onReceive(method: String, params: List, json: ByteArray) { try { if (method == "eth_getTransactionReceipt") { cacheTxReceipt(params, json) @@ -42,7 +43,7 @@ class CacheRequested( } } - fun cacheTxReceipt(params: List, json: ByteArray) { + fun cacheTxReceipt(params: List, json: ByteArray) { if (params.size != 1) { return } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy index 5baaa092..fa1d0a03 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy @@ -19,12 +19,17 @@ import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global 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.test.TestingCommons +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId 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 reactor.core.publisher.Mono import spock.lang.Specification @@ -227,4 +232,36 @@ class CachesSpec extends Specification { 1 * blocksCache.read(block.hash) >> Mono.just(block) 1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then() } + + def "Put receipt into mem cache"() { + setup: + def receipt = new TransactionReceiptJson().tap { + transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1") + blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b") + blockNumber = 0xccf6e2 + from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a") + logs = [] + } + def receiptContainer = new DefaultContainer( + TxId.from(receipt.transactionHash), + BlockId.from(receipt.blockHash), + receipt.blockNumber, + Global.objectMapper.writeValueAsBytes(receipt), + receipt + ) + + ReceiptMemCache receiptMemCache = Mock() + def caches = Caches.newBuilder() + .setReceipts(receiptMemCache) + .build() + Head head = Mock() + caches.setHead(head) + when: + caches.cacheReceipt(Caches.Tag.REQUESTED, receiptContainer) + + then: + 1 * head.getCurrentHeight() >> 0xccf6e2 + 1 * receiptMemCache.acceptsRecentBlocks(0) >> true + 1 * receiptMemCache.add(receiptContainer) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy new file mode 100644 index 00000000..01c88ba7 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy @@ -0,0 +1,100 @@ +/** + * 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.cache + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.DefaultContainer +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson +import spock.lang.Specification + +import java.time.Instant + +class ReceiptMemCacheSpec extends Specification { + + ObjectMapper objectMapper = Global.objectMapper + + def "Add and read"() { + setup: + def cache = new ReceiptMemCache() + + def receipt = new TransactionReceiptJson().tap { + transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1") + blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b") + blockNumber = 0xccf6e2 + from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a") + logs = [] + } + def receiptContainer = new DefaultContainer( + TxId.from(receipt.transactionHash), + BlockId.from(receipt.blockHash), + receipt.blockNumber, + objectMapper.writeValueAsBytes(receipt), + receipt + ) + + when: + cache.add(receiptContainer) + def act = cache.read(TxId.from(receipt.transactionHash)).block() + then: + act != null + objectMapper.readValue(act, TransactionReceiptJson.class) == receipt + } + + def "Evict by block"() { + setup: + def cache = new ReceiptMemCache() + + def receipt = new TransactionReceiptJson().tap { + transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1") + blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b") + blockNumber = 0xccf6e2 + from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a") + logs = [] + } + def receiptContainer = new DefaultContainer( + TxId.from(receipt.transactionHash), + BlockId.from(receipt.blockHash), + receipt.blockNumber, + objectMapper.writeValueAsBytes(receipt), + receipt + ) + + def blockContainer = new BlockContainer( + receipt.blockNumber, BlockId.from(receipt.blockHash), + BigInteger.ONE, + Instant.now(), + false, + "{}".bytes, + null, + [TxId.from(receipt.transactionHash)] + ) + + when: + cache.add(receiptContainer) + cache.evict(blockContainer) + def act = cache.read(TxId.from(receipt.transactionHash)).block() + then: + act == null + } + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index b2e8f08f..78a047d5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -17,12 +17,18 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.quorum.AlwaysQuorum +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain +import org.jetbrains.annotations.NotNull +import reactor.core.publisher.Mono import spock.lang.Specification import java.time.Duration @@ -163,4 +169,61 @@ class MultistreamSpec extends Specification { then: !act } + + def "Call postprocess after api use"() { + setup: + def request = new JsonRpcRequest("test_foo", [1], 1) + + 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) + } + + class TestMultistream extends Multistream { + + TestMultistream(List upstreams, @NotNull RequestPostprocessor postprocessor) { + super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor) + } + + @Override + Mono> 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 getLabels() { + return null + } + + public T cast(Class selfType) { + return this + } + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy new file mode 100644 index 00000000..5d298f4f --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy @@ -0,0 +1,47 @@ +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) + 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) + def processor = Mock(RequestPostprocessor) + Reader 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(_, _, _) + } +}